From 3b978b40704e43480109c2655d717aa08d2b782e Mon Sep 17 00:00:00 2001 From: Nitesh Purohit Date: Wed, 9 Sep 2026 17:01:20 -0400 Subject: [PATCH 1/5] test: Introduce E2ESubprocess for better process handling - Added E2ESubprocess class to manage subprocess execution and output capturing. - Replaced Open3 calls with E2ESubprocess in CLI worker specs for consistency and improved error handling. - Updated framework runtime control support to utilize E2ESubprocess for better process management. - Introduced tests for E2ESubprocess to ensure proper timeout handling and output capturing. --- .../spec/e2e/karya/cli_worker_mysql_spec.rb | 3 +- .../spec/e2e/karya/cli_worker_redis_spec.rb | 3 +- gems/karya/spec/e2e/karya/cli_worker_spec.rb | 48 +++---- .../spec/e2e/karya/cli_worker_sqlite_spec.rb | 3 +- .../spec/e2e/karya/e2_e_subprocess_spec.rb | 28 ++++ spec/support/e2e_subprocess.rb | 121 ++++++++++++++++++ .../framework_runtime_control_e2e_support.rb | 101 ++++++++------- 7 files changed, 224 insertions(+), 83 deletions(-) create mode 100644 gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb create mode 100644 spec/support/e2e_subprocess.rb diff --git a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb index ef96334a..075e7fb2 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb @@ -7,6 +7,7 @@ require_relative '../spec_helper' require File.expand_path('../../../../../spec/support/mysql_e2e_support', __dir__) +require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) RSpec.describe Karya::CLI, :e2e, :integration do def mysql_boot_file(mysql_url:, namespace:, marker_file:) @@ -52,7 +53,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, mysql_boot_file(mysql_url: database_url, namespace:, marker_file:)) - stdout, stderr, status = Open3.capture3( + stdout, stderr, status = E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb index bde6fff9..8b2f60d3 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb @@ -6,6 +6,7 @@ # LICENSE file in the root directory of this source tree. require_relative '../spec_helper' +require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) RSpec.describe Karya::CLI, :e2e, :integration do def delete_redis_namespace(redis_url:, namespace:) @@ -58,7 +59,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, redis_boot_file(redis_url:, namespace:, marker_file:)) - stdout, stderr, status = Open3.capture3( + stdout, stderr, status = E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/cli_worker_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_spec.rb index 5aa3b727..15ca548e 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_spec.rb @@ -6,10 +6,11 @@ # LICENSE file in the root directory of this source tree. require_relative '../spec_helper' +require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) RSpec.describe Karya::CLI, :e2e, :integration do def run_cli(*args) - Open3.capture3(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT) + E2ESubprocess.capture(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT) end def wait_for_runtime_phase(state_file, *phases) @@ -41,7 +42,7 @@ def build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:) end def with_force_stop_worker(boot_file:, state_file:, &) - Open3.popen2e(*karya_command( + process = E2ESubprocess.new(*karya_command( 'worker', 'billing', '--require', @@ -58,7 +59,10 @@ def with_force_stop_worker(boot_file:, state_file:, &) '0', '--state-file', state_file - ), chdir: KaryaE2EHelpers::PACKAGE_ROOT, &) + ), chdir: KaryaE2EHelpers::PACKAGE_ROOT) + yield process + ensure + process&.close end def request_force_stop(supervisor_pid:, draining_marker_file:) @@ -69,39 +73,23 @@ def request_force_stop(supervisor_pid:, draining_marker_file:) nil end - def cleanup_worker_process(wait_thr) - return unless wait_thr.alive? - - Process.kill('TERM', wait_thr.pid) - sleep(0.1) - Process.kill('KILL', wait_thr.pid) if wait_thr.alive? - rescue Errno::ESRCH - nil - end - - def expect_force_stopped_worker(wait_thr:, stdout_and_stderr:, state_file:, marker_file:) - process_status = Timeout.timeout(10) { wait_thr.value } - output = stdout_and_stderr.read + def expect_force_stopped_worker(process:, state_file:, marker_file:) + process_status = process.wait(timeout: 10) + process.wait_for_output runtime_state = wait_for_runtime_phase(state_file, 'stopped', 'force_stopping') - expect(process_status.exitstatus).to eq(1), -> { "worker output:\n#{output}" } + expect(process_status.exitstatus).to eq(1), -> { "worker output:\n#{process.output}" } expect(File.read(marker_file).strip).not_to be_empty expect(runtime_state.fetch('phase')).to match(/\A(?:stopped|force_stopping)\z/) end - def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_file:) + def with_force_stop_cleanup(process:, state_file:, marker_file:) yield expect_force_stopped_worker( - wait_thr:, - stdout_and_stderr:, + process:, state_file:, marker_file: ) - rescue Timeout::Error, RSpec::Expectations::ExpectationNotMetError - cleanup_worker_process(wait_thr) - raise - ensure - cleanup_worker_process(wait_thr) end it 'executes a queued job end-to-end through exe/karya worker' do @@ -154,10 +142,9 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f draining_marker_file = File.join(directory, 'draining.txt') boot_file = build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:) - with_force_stop_worker(boot_file:, state_file:) do |_stdin, stdout_and_stderr, wait_thr| + with_force_stop_worker(boot_file:, state_file:) do |process| with_force_stop_cleanup( - wait_thr:, - stdout_and_stderr:, + process:, state_file:, marker_file: ) do @@ -176,7 +163,7 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f draining_marker_file = File.join(directory, 'draining.txt') boot_file = build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:) - with_force_stop_worker(boot_file:, state_file:) do |_stdin, stdout_and_stderr, wait_thr| + with_force_stop_worker(boot_file:, state_file:) do |process| expect(wait_until { File.exist?(marker_file) && File.exist?(state_file) }).to be(true) inspect_stdout, inspect_stderr, inspect_status = run_cli('runtime', 'inspect', '--state-file', state_file) @@ -195,8 +182,7 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" } runtime_phase = wait_for_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase') expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/) - cleanup_worker_process(wait_thr) - stdout_and_stderr.read + process.close end end end diff --git a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb index 165e057d..5e23131a 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb @@ -7,6 +7,7 @@ require_relative '../spec_helper' require File.expand_path('../../../../../spec/support/sqlite_e2e_support', __dir__) +require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) RSpec.describe Karya::CLI, :e2e, :integration do def sqlite_boot_file(sqlite_url:, namespace:, marker_file:) @@ -50,7 +51,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, sqlite_boot_file(sqlite_url: database_url, namespace:, marker_file:)) - stdout, stderr, status = Open3.capture3( + stdout, stderr, status = E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb b/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb new file mode 100644 index 00000000..fe154572 --- /dev/null +++ b/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb @@ -0,0 +1,28 @@ +# frozen_string_literal: true + +# Copyright Codevedas Inc. 2025-present +# +# This source code is licensed under the MIT license found in the +# LICENSE file in the root directory of this source tree. + +require_relative '../spec_helper' +require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) + +RSpec.describe E2ESubprocess, :e2e, :integration do + it 'bounds capture when a forked descendant retains an output pipe' do + command = [ + RbConfig.ruby, + '-e', + <<~RUBY + fork do + Signal.trap('TERM', 'IGNORE') + loop { sleep 0.1 } + end + RUBY + ] + + expect do + described_class.capture(*command, timeout: 0.5) + end.to raise_error(Timeout::Error, /subprocess timed out after 0.5s/) + end +end diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb new file mode 100644 index 00000000..1a07e6ea --- /dev/null +++ b/spec/support/e2e_subprocess.rb @@ -0,0 +1,121 @@ +# frozen_string_literal: true + +# Copyright Codevedas Inc. 2025-present +# +# This source code is licensed under the MIT license found in the +# LICENSE file in the root directory of this source tree. + +require 'open3' +require 'timeout' + +class E2ESubprocess + DEFAULT_TIMEOUT = 30 + TERMINATION_TIMEOUT = 2 + + attr_reader :pid + + def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) + process = new(*command, **options) + status = process.wait(timeout:) + process.wait_for_output(timeout: TERMINATION_TIMEOUT) + [process.stdout, process.stderr, status] + rescue Timeout::Error + raise Timeout::Error, "subprocess timed out after #{timeout}s:\n#{process&.output}" + ensure + process&.close + end + + def initialize(*command, **options) + @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options, pgroup: true) + @pid = @wait_thread.pid + @stdout = +'' + @stderr = +'' + @output_lock = Mutex.new + @stdout_reader = read_output(@stdout_io, @stdout) + @stderr_reader = read_output(@stderr_io, @stderr) + @stdin.close + end + + def alive? + @wait_thread.alive? + end + + def wait(timeout: DEFAULT_TIMEOUT) + Timeout.timeout(timeout) { @wait_thread.value } + end + + def wait_for_output(timeout: TERMINATION_TIMEOUT) + Timeout.timeout(timeout) do + @stdout_reader.value + @stderr_reader.value + end + end + + def stdout + output_snapshot(@stdout) + end + + def stderr + output_snapshot(@stderr) + end + + def output + [stdout, stderr].reject(&:empty?).join("\n") + end + + def close + terminate_process_group if alive? || output_readers_alive? + close_output_streams + reap_process + reap_output_readers + end + + private + + def read_output(stream, destination) + Thread.new do + loop do + chunk = stream.readpartial(4096) + @output_lock.synchronize { destination << chunk } + end + rescue EOFError, IOError + nil + end + end + + def output_snapshot(buffer) + @output_lock.synchronize { buffer.dup } + end + + def output_readers_alive? + @stdout_reader.alive? || @stderr_reader.alive? + end + + def terminate_process_group + signal_process_group('TERM') + @wait_thread.join(TERMINATION_TIMEOUT) + return unless alive? || output_readers_alive? + + signal_process_group('KILL') + end + + def signal_process_group(signal) + Process.kill(signal, -pid) + rescue Errno::ESRCH + nil + end + + def close_output_streams + @stdout_io.close unless @stdout_io.closed? + @stderr_io.close unless @stderr_io.closed? + end + + def reap_process + @wait_thread.join(TERMINATION_TIMEOUT) + end + + def reap_output_readers + @stdout_reader.join(TERMINATION_TIMEOUT) + @stderr_reader.join(TERMINATION_TIMEOUT) + end +end diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb index 4ccc12b7..28b25049 100644 --- a/spec/support/framework_runtime_control_e2e_support.rb +++ b/spec/support/framework_runtime_control_e2e_support.rb @@ -12,6 +12,7 @@ require 'securerandom' require 'timeout' require 'tmpdir' +require_relative 'e2e_subprocess' module FrameworkRuntimeControlE2ESupport def framework_runtime_control_env(framework:, framework_gem_root:, database_url:, namespace:, worker_name:, worker: false) @@ -165,11 +166,11 @@ def wait_for_framework_runtime_phase(state_file, *phases) end end - def wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr) + def wait_for_framework_runtime_start(state_file, process) wait_until do - if wait_thr.join(0) - output = stdout_and_stderr.read - raise "worker exited before runtime control started:\n#{output}" + unless process.alive? + process.wait_for_output + raise "worker exited before runtime control started:\n#{process.output}" end next unless File.exist?(state_file) @@ -204,18 +205,12 @@ def wait_until(timeout: 10) end end - def cleanup_framework_worker_process(wait_thr) - return unless wait_thr.alive? - - Process.kill('TERM', wait_thr.pid) - Timeout.timeout(2) do - wait_thr.join - end - Process.kill('KILL', wait_thr.pid) if wait_thr.alive? - rescue Timeout::Error - Process.kill('KILL', wait_thr.pid) if wait_thr.alive? - rescue Errno::ESRCH - nil + def run_framework_runtime_command(framework:, action:, queue:, worker_name:, env:) + E2ESubprocess.capture( + env, + *framework_runtime_command(framework:, action:, queue:, worker_name:), + chdir: current_app_root + ) end def request_framework_force_stop(supervisor_pid:, draining_marker_path:) @@ -316,68 +311,75 @@ def runtime_control_rakefile_source(framework:) worker: true ) - Open3.popen2e( + process = E2ESubprocess.new( command_env, *framework_worker_command(framework:, queue:, worker_name:), chdir: current_app_root - ) do |_stdin, stdout_and_stderr, wait_thr| - begin - wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr) - - inspect_stdout, inspect_stderr, inspect_status = Open3.capture3( + ) + begin + wait_for_framework_runtime_start(state_file, process) + + inspect_stdout, inspect_stderr, inspect_status = run_framework_runtime_command( + framework:, + action: :inspect, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, framework_gem_root:, database_url:, namespace:, worker_name: - ), - *framework_runtime_command(framework:, action: :inspect, queue:, worker_name:), - chdir: current_app_root + ) ) inspect_payload = parse_framework_runtime_json(inspect_stdout) expect(inspect_status.exitstatus).to eq(0), -> { "stderr:\n#{inspect_stderr}" } expect(inspect_payload.fetch('snapshot').fetch('phase')).to match(/\A(?:starting|running|draining)\z/) - drain_stdout, drain_stderr, drain_status = Open3.capture3( + drain_stdout, drain_stderr, drain_status = run_framework_runtime_command( + framework:, + action: :drain, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, framework_gem_root:, database_url:, namespace:, worker_name: - ), - *framework_runtime_command(framework:, action: :drain, queue:, worker_name:), - chdir: current_app_root + ) ) expect(drain_status.exitstatus).to eq(0), -> { "stdout:\n#{drain_stdout}\n\nstderr:\n#{drain_stderr}" } expect(wait_for_framework_runtime_phase(state_file, 'draining')).to include('phase' => 'draining') - force_stop_stdout, force_stop_stderr, force_stop_status = Open3.capture3( + force_stop_stdout, force_stop_stderr, force_stop_status = run_framework_runtime_command( + framework:, + action: :force_stop, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, framework_gem_root:, database_url:, namespace:, worker_name: - ), - *framework_runtime_command(framework:, action: :force_stop, queue:, worker_name:), - chdir: current_app_root + ) ) expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" } runtime_phase = wait_for_framework_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase') expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/) rescue Timeout::Error - output = wait_thr.join(0) ? stdout_and_stderr.read : '(worker still running)' state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' - raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{output}" + raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" ensure - cleanup_framework_worker_process(wait_thr) + process.close end - end end end @@ -400,25 +402,28 @@ def runtime_control_rakefile_source(framework:) worker: true ) - Open3.popen2e( + process = E2ESubprocess.new( worker_env, *framework_worker_command(framework:, queue:, worker_name:), chdir: current_app_root - ) do |_stdin, stdout_and_stderr, wait_thr| - begin - wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr) + ) + begin + wait_for_framework_runtime_start(state_file, process) write_stale_runtime_state_file!(live_state_file: state_file, stale_state_file:) - command_stdout, command_stderr, command_status = Open3.capture3( + command_stdout, command_stderr, command_status = run_framework_runtime_command( + framework:, + action: :drain, + queue:, + worker_name: stale_worker_name, + env: framework_runtime_control_env( framework:, framework_gem_root:, database_url:, namespace:, worker_name: stale_worker_name - ), - *framework_runtime_command(framework:, action: :drain, queue:, worker_name: stale_worker_name), - chdir: current_app_root + ) ) combined_output = "#{command_stdout}\n#{command_stderr}" @@ -438,14 +443,12 @@ def runtime_control_rakefile_source(framework:) } expect(combined_output).to include('runtime control token does not match the running supervisor') rescue Timeout::Error - output = wait_thr.join(0) ? stdout_and_stderr.read : '(worker still running)' state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' - raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{output}" + raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" ensure File.delete(stale_state_file) if File.exist?(stale_state_file) - cleanup_framework_worker_process(wait_thr) + process.close end - end end end end From c422f888e43a0ae94679c58dabebaeaff8e27490 Mon Sep 17 00:00:00 2001 From: Nitesh Purohit Date: Wed, 9 Sep 2026 19:15:13 -0400 Subject: [PATCH 2/5] test: Enhance E2ESubprocess timeout handling - Introduced a deadline mechanism in E2ESubprocess to manage timeouts more effectively. - Updated the capture method to use remaining time for subprocess execution. - Wrapped the capture call in a Timeout block to ensure proper error handling. - Improved test case to reflect the new timeout behavior for subprocesses. --- .../spec/e2e/karya/e2_e_subprocess_spec.rb | 4 +- spec/support/e2e_subprocess.rb | 20 +- .../framework_runtime_control_e2e_support.rb | 187 +++++++++--------- 3 files changed, 111 insertions(+), 100 deletions(-) diff --git a/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb b/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb index fe154572..7424fe97 100644 --- a/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb +++ b/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb @@ -22,7 +22,9 @@ ] expect do - described_class.capture(*command, timeout: 0.5) + Timeout.timeout(5) do + described_class.capture(*command, timeout: 0.5) + end end.to raise_error(Timeout::Error, /subprocess timed out after 0.5s/) end end diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb index 1a07e6ea..bc2fc660 100644 --- a/spec/support/e2e_subprocess.rb +++ b/spec/support/e2e_subprocess.rb @@ -15,9 +15,10 @@ class E2ESubprocess attr_reader :pid def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) + deadline = monotonic_time + timeout process = new(*command, **options) - status = process.wait(timeout:) - process.wait_for_output(timeout: TERMINATION_TIMEOUT) + status = process.wait(timeout: remaining_time(deadline)) + process.wait_for_output(timeout: remaining_time(deadline)) [process.stdout, process.stderr, status] rescue Timeout::Error raise Timeout::Error, "subprocess timed out after #{timeout}s:\n#{process&.output}" @@ -25,8 +26,21 @@ def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) process&.close end + def self.monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + private_class_method :monotonic_time + + def self.remaining_time(deadline) + remaining = deadline - monotonic_time + raise Timeout::Error unless remaining.positive? + + remaining + end + private_class_method :remaining_time + def initialize(*command, **options) - @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options, pgroup: true) + @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options.merge(pgroup: true)) @pid = @wait_thread.pid @stdout = +'' @stderr = +'' diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb index 28b25049..b4a174b2 100644 --- a/spec/support/framework_runtime_control_e2e_support.rb +++ b/spec/support/framework_runtime_control_e2e_support.rb @@ -7,7 +7,6 @@ require 'json' require 'fileutils' -require 'open3' require 'rbconfig' require 'securerandom' require 'timeout' @@ -307,8 +306,8 @@ def runtime_control_rakefile_source(framework:) framework_gem_root:, database_url:, namespace:, - worker_name:, - worker: true + worker_name:, + worker: true ) process = E2ESubprocess.new( @@ -317,69 +316,66 @@ def runtime_control_rakefile_source(framework:) chdir: current_app_root ) begin - wait_for_framework_runtime_start(state_file, process) + wait_for_framework_runtime_start(state_file, process) - inspect_stdout, inspect_stderr, inspect_status = run_framework_runtime_command( + inspect_stdout, inspect_stderr, inspect_status = run_framework_runtime_command( + framework:, + action: :inspect, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, - action: :inspect, - queue:, - worker_name:, - env: - framework_runtime_control_env( - framework:, - framework_gem_root:, - database_url:, - namespace:, - worker_name: - ) + framework_gem_root:, + database_url:, + namespace:, + worker_name: ) - inspect_payload = parse_framework_runtime_json(inspect_stdout) + ) + inspect_payload = parse_framework_runtime_json(inspect_stdout) - expect(inspect_status.exitstatus).to eq(0), -> { "stderr:\n#{inspect_stderr}" } - expect(inspect_payload.fetch('snapshot').fetch('phase')).to match(/\A(?:starting|running|draining)\z/) + expect(inspect_status.exitstatus).to eq(0), -> { "stderr:\n#{inspect_stderr}" } + expect(inspect_payload.fetch('snapshot').fetch('phase')).to match(/\A(?:starting|running|draining)\z/) - drain_stdout, drain_stderr, drain_status = run_framework_runtime_command( + drain_stdout, drain_stderr, drain_status = run_framework_runtime_command( + framework:, + action: :drain, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, - action: :drain, - queue:, - worker_name:, - env: - framework_runtime_control_env( - framework:, - framework_gem_root:, - database_url:, - namespace:, - worker_name: - ) + framework_gem_root:, + database_url:, + namespace:, + worker_name: ) + ) - expect(drain_status.exitstatus).to eq(0), -> { "stdout:\n#{drain_stdout}\n\nstderr:\n#{drain_stderr}" } - expect(wait_for_framework_runtime_phase(state_file, 'draining')).to include('phase' => 'draining') + expect(drain_status.exitstatus).to eq(0), -> { "stdout:\n#{drain_stdout}\n\nstderr:\n#{drain_stderr}" } + expect(wait_for_framework_runtime_phase(state_file, 'draining')).to include('phase' => 'draining') - force_stop_stdout, force_stop_stderr, force_stop_status = run_framework_runtime_command( + force_stop_stdout, force_stop_stderr, force_stop_status = run_framework_runtime_command( + framework:, + action: :force_stop, + queue:, + worker_name:, + env: framework_runtime_control_env( framework:, - action: :force_stop, - queue:, - worker_name:, - env: - framework_runtime_control_env( - framework:, - framework_gem_root:, - database_url:, - namespace:, - worker_name: - ) + framework_gem_root:, + database_url:, + namespace:, + worker_name: ) - - expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" } - runtime_phase = wait_for_framework_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase') - expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/) - rescue Timeout::Error - state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' - raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" - ensure - process.close - end + ) + + expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" } + runtime_phase = wait_for_framework_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase') + expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/) + rescue Timeout::Error + state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' + raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" + ensure + process.close + end end end @@ -398,8 +394,8 @@ def runtime_control_rakefile_source(framework:) framework_gem_root:, database_url:, namespace:, - worker_name:, - worker: true + worker_name:, + worker: true ) process = E2ESubprocess.new( @@ -408,47 +404,46 @@ def runtime_control_rakefile_source(framework:) chdir: current_app_root ) begin - wait_for_framework_runtime_start(state_file, process) - write_stale_runtime_state_file!(live_state_file: state_file, stale_state_file:) - - command_stdout, command_stderr, command_status = run_framework_runtime_command( + wait_for_framework_runtime_start(state_file, process) + write_stale_runtime_state_file!(live_state_file: state_file, stale_state_file:) + + command_stdout, command_stderr, command_status = run_framework_runtime_command( + framework:, + action: :drain, + queue:, + worker_name: stale_worker_name, + env: framework_runtime_control_env( framework:, - action: :drain, - queue:, - worker_name: stale_worker_name, - env: - framework_runtime_control_env( - framework:, - framework_gem_root:, - database_url:, - namespace:, - worker_name: stale_worker_name - ) + framework_gem_root:, + database_url:, + namespace:, + worker_name: stale_worker_name ) - - combined_output = "#{command_stdout}\n#{command_stderr}" - expect(command_status.exitstatus).not_to eq(0), lambda { - stale_state_payload = File.exist?(stale_state_file) ? File.read(stale_state_file) : '(missing state file)' - <<~TEXT - expected stale-token runtime command to fail - stdout: - #{command_stdout} - - stderr: - #{command_stderr} - - state: - #{stale_state_payload} - TEXT - } - expect(combined_output).to include('runtime control token does not match the running supervisor') - rescue Timeout::Error - state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' - raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" - ensure - File.delete(stale_state_file) if File.exist?(stale_state_file) - process.close - end + ) + + combined_output = "#{command_stdout}\n#{command_stderr}" + expect(command_status.exitstatus).not_to eq(0), lambda { + stale_state_payload = File.exist?(stale_state_file) ? File.read(stale_state_file) : '(missing state file)' + <<~TEXT + expected stale-token runtime command to fail + stdout: + #{command_stdout} + + stderr: + #{command_stderr} + + state: + #{stale_state_payload} + TEXT + } + expect(combined_output).to include('runtime control token does not match the running supervisor') + rescue Timeout::Error + state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' + raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" + ensure + File.delete(stale_state_file) if File.exist?(stale_state_file) + process.close + end end end end From 0bdcb110dad055d0e3287380f9e45411e83c9bbc Mon Sep 17 00:00:00 2001 From: Nitesh Purohit Date: Wed, 9 Sep 2026 19:30:56 -0400 Subject: [PATCH 3/5] refactor: Update E2ESubprocess references to KaryaSpecSupport --- .../spec/e2e/karya/cli_worker_mysql_spec.rb | 2 +- .../spec/e2e/karya/cli_worker_redis_spec.rb | 2 +- gems/karya/spec/e2e/karya/cli_worker_spec.rb | 13 +- .../spec/e2e/karya/cli_worker_sqlite_spec.rb | 2 +- .../e2_e_subprocess_spec.rb | 2 +- spec/support/e2e_subprocess.rb | 249 +++++++++++------- .../framework_runtime_control_e2e_support.rb | 8 +- 7 files changed, 167 insertions(+), 111 deletions(-) rename gems/karya/spec/e2e/{karya => karya_spec_support}/e2_e_subprocess_spec.rb (91%) diff --git a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb index 075e7fb2..e71cddcb 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb @@ -53,7 +53,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, mysql_boot_file(mysql_url: database_url, namespace:, marker_file:)) - stdout, stderr, status = E2ESubprocess.capture( + stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb index 8b2f60d3..e7aa8545 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb @@ -59,7 +59,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, redis_boot_file(redis_url:, namespace:, marker_file:)) - stdout, stderr, status = E2ESubprocess.capture( + stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/cli_worker_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_spec.rb index 15ca548e..2213888f 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_spec.rb @@ -10,7 +10,7 @@ RSpec.describe Karya::CLI, :e2e, :integration do def run_cli(*args) - E2ESubprocess.capture(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT) + KaryaSpecSupport::E2ESubprocess.capture(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT) end def wait_for_runtime_phase(state_file, *phases) @@ -42,7 +42,7 @@ def build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:) end def with_force_stop_worker(boot_file:, state_file:, &) - process = E2ESubprocess.new(*karya_command( + process = KaryaSpecSupport::E2ESubprocess.new(*karya_command( 'worker', 'billing', '--require', @@ -75,7 +75,11 @@ def request_force_stop(supervisor_pid:, draining_marker_file:) def expect_force_stopped_worker(process:, state_file:, marker_file:) process_status = process.wait(timeout: 10) - process.wait_for_output + begin + process.wait_for_output + rescue Timeout::Error + raise Timeout::Error, "force-stopped worker output remained open:\n#{process.output}" + end runtime_state = wait_for_runtime_phase(state_file, 'stopped', 'force_stopping') expect(process_status.exitstatus).to eq(1), -> { "worker output:\n#{process.output}" } @@ -163,7 +167,7 @@ def with_force_stop_cleanup(process:, state_file:, marker_file:) draining_marker_file = File.join(directory, 'draining.txt') boot_file = build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:) - with_force_stop_worker(boot_file:, state_file:) do |process| + with_force_stop_worker(boot_file:, state_file:) do expect(wait_until { File.exist?(marker_file) && File.exist?(state_file) }).to be(true) inspect_stdout, inspect_stderr, inspect_status = run_cli('runtime', 'inspect', '--state-file', state_file) @@ -182,7 +186,6 @@ def with_force_stop_cleanup(process:, state_file:, marker_file:) expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" } runtime_phase = wait_for_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase') expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/) - process.close end end end diff --git a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb index 5e23131a..cd617a78 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb @@ -51,7 +51,7 @@ def self.call(account_id:, marker_path:) File.write(boot_file, sqlite_boot_file(sqlite_url: database_url, namespace:, marker_file:)) - stdout, stderr, status = E2ESubprocess.capture( + stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture( *karya_command( 'worker', 'billing', diff --git a/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb similarity index 91% rename from gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb rename to gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb index 7424fe97..5581d2f3 100644 --- a/gems/karya/spec/e2e/karya/e2_e_subprocess_spec.rb +++ b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb @@ -8,7 +8,7 @@ require_relative '../spec_helper' require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) -RSpec.describe E2ESubprocess, :e2e, :integration do +RSpec.describe KaryaSpecSupport::E2ESubprocess, :e2e, :integration do it 'bounds capture when a forked descendant retains an output pipe' do command = [ RbConfig.ruby, diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb index bc2fc660..0e42fc5b 100644 --- a/spec/support/e2e_subprocess.rb +++ b/spec/support/e2e_subprocess.rb @@ -8,128 +8,179 @@ require 'open3' require 'timeout' -class E2ESubprocess - DEFAULT_TIMEOUT = 30 - TERMINATION_TIMEOUT = 2 - - attr_reader :pid - - def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) - deadline = monotonic_time + timeout - process = new(*command, **options) - status = process.wait(timeout: remaining_time(deadline)) - process.wait_for_output(timeout: remaining_time(deadline)) - [process.stdout, process.stderr, status] - rescue Timeout::Error - raise Timeout::Error, "subprocess timed out after #{timeout}s:\n#{process&.output}" - ensure - process&.close - end +module KaryaSpecSupport + class E2ESubprocess + class CleanupError < StandardError; end + + DEFAULT_TIMEOUT = 30 + TERMINATION_TIMEOUT = 2 + + attr_reader :pid + + def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) + deadline = monotonic_time + timeout + cleanup_reserve = [TERMINATION_TIMEOUT, timeout / 2.0].min + operation_deadline = deadline - cleanup_reserve + process = new(*command, **options) + begin + status = process.wait(timeout: remaining_time(operation_deadline)) + process.wait_for_output(timeout: remaining_time(operation_deadline)) + result = [process.stdout, process.stderr, status] + rescue Timeout::Error + failure = Timeout::Error.new("subprocess timed out after #{timeout}s:\n#{process.output}") + ensure + begin + process.close(deadline:) + rescue CleanupError => error + failure = failure ? CleanupError.new("#{failure.message}\n#{error.message}") : error + end + end - def self.monotonic_time - Process.clock_gettime(Process::CLOCK_MONOTONIC) - end - private_class_method :monotonic_time + raise failure if failure - def self.remaining_time(deadline) - remaining = deadline - monotonic_time - raise Timeout::Error unless remaining.positive? + result + end - remaining - end - private_class_method :remaining_time - - def initialize(*command, **options) - @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options.merge(pgroup: true)) - @pid = @wait_thread.pid - @stdout = +'' - @stderr = +'' - @output_lock = Mutex.new - @stdout_reader = read_output(@stdout_io, @stdout) - @stderr_reader = read_output(@stderr_io, @stderr) - @stdin.close - end + def self.monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + private_class_method :monotonic_time - def alive? - @wait_thread.alive? - end + def self.remaining_time(deadline) + remaining = deadline - monotonic_time + raise Timeout::Error unless remaining.positive? - def wait(timeout: DEFAULT_TIMEOUT) - Timeout.timeout(timeout) { @wait_thread.value } - end + remaining + end + private_class_method :remaining_time + + def initialize(*command, **options) + @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options.merge(pgroup: true)) + @pid = @wait_thread.pid + @stdout = +'' + @stderr = +'' + @output_lock = Mutex.new + @stdout_reader = read_output(@stdout_io, @stdout) + @stderr_reader = read_output(@stderr_io, @stderr) + @stdin.close + end - def wait_for_output(timeout: TERMINATION_TIMEOUT) - Timeout.timeout(timeout) do - @stdout_reader.value - @stderr_reader.value + def alive? + @wait_thread.alive? end - end - def stdout - output_snapshot(@stdout) - end + def wait(timeout: DEFAULT_TIMEOUT) + Timeout.timeout(timeout) { @wait_thread.value } + end - def stderr - output_snapshot(@stderr) - end + def wait_for_output(timeout: TERMINATION_TIMEOUT) + Timeout.timeout(timeout) do + @stdout_reader.value + @stderr_reader.value + end + end - def output - [stdout, stderr].reject(&:empty?).join("\n") - end + def stdout + output_snapshot(@stdout) + end - def close - terminate_process_group if alive? || output_readers_alive? - close_output_streams - reap_process - reap_output_readers - end + def stderr + output_snapshot(@stderr) + end + + def output + [stdout, stderr].reject(&:empty?).join("\n") + end - private + def close(deadline: monotonic_time + TERMINATION_TIMEOUT) + terminate_process_group(deadline) if alive? || output_readers_alive? + close_output_streams + join_until(@wait_thread, deadline) + join_until(@stdout_reader, deadline) + join_until(@stderr_reader, deadline) + wait_for_shutdown(deadline) + verify_closed + end + + private - def read_output(stream, destination) - Thread.new do - loop do - chunk = stream.readpartial(4096) - @output_lock.synchronize { destination << chunk } + def read_output(stream, destination) + Thread.new do + loop do + chunk = stream.readpartial(4096) + @output_lock.synchronize { destination << chunk } + end + rescue EOFError, IOError + nil end - rescue EOFError, IOError + end + + def output_snapshot(buffer) + @output_lock.synchronize { buffer.dup } + end + + def output_readers_alive? + @stdout_reader.alive? || @stderr_reader.alive? + end + + def terminate_process_group(deadline) + signal_process_group('TERM') + grace_deadline = monotonic_time + (remaining_time(deadline) / 2.0) + join_until(@wait_thread, grace_deadline) + return unless alive? || output_readers_alive? + + signal_process_group('KILL') + end + + def signal_process_group(signal) + Process.kill(signal, -pid) + rescue Errno::ESRCH nil end - end - def output_snapshot(buffer) - @output_lock.synchronize { buffer.dup } - end + def close_output_streams + @stdout_io.close unless @stdout_io.closed? + @stderr_io.close unless @stderr_io.closed? + end - def output_readers_alive? - @stdout_reader.alive? || @stderr_reader.alive? - end + def join_until(thread, deadline) + thread.join([remaining_time(deadline), 0].max) + end - def terminate_process_group - signal_process_group('TERM') - @wait_thread.join(TERMINATION_TIMEOUT) - return unless alive? || output_readers_alive? + def verify_closed + return if shutdown_complete? - signal_process_group('KILL') - end + raise CleanupError, "subprocess cleanup incomplete for process group #{pid}:\n#{output}" + end - def signal_process_group(signal) - Process.kill(signal, -pid) - rescue Errno::ESRCH - nil - end + def wait_for_shutdown(deadline) + until shutdown_complete? + remaining = remaining_time(deadline) + break unless remaining.positive? - def close_output_streams - @stdout_io.close unless @stdout_io.closed? - @stderr_io.close unless @stderr_io.closed? - end + sleep([remaining, 0.01].min) + end + end - def reap_process - @wait_thread.join(TERMINATION_TIMEOUT) - end + def shutdown_complete? + !alive? && !output_readers_alive? && !process_group_alive? + end + + def process_group_alive? + Process.kill(0, -pid) + true + rescue Errno::ESRCH + false + rescue Errno::EPERM + true + end - def reap_output_readers - @stdout_reader.join(TERMINATION_TIMEOUT) - @stderr_reader.join(TERMINATION_TIMEOUT) + def monotonic_time + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def remaining_time(deadline) + deadline - monotonic_time + end end end diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb index b4a174b2..b182ba9e 100644 --- a/spec/support/framework_runtime_control_e2e_support.rb +++ b/spec/support/framework_runtime_control_e2e_support.rb @@ -183,6 +183,8 @@ def wait_for_framework_runtime_start(state_file, process) payload rescue Errno::ENOENT, JSON::ParserError, KeyError next + rescue Timeout::Error + raise "worker exited before runtime control started and output remained open:\n#{process.output}" end end @@ -205,7 +207,7 @@ def wait_until(timeout: 10) end def run_framework_runtime_command(framework:, action:, queue:, worker_name:, env:) - E2ESubprocess.capture( + KaryaSpecSupport::E2ESubprocess.capture( env, *framework_runtime_command(framework:, action:, queue:, worker_name:), chdir: current_app_root @@ -310,7 +312,7 @@ def runtime_control_rakefile_source(framework:) worker: true ) - process = E2ESubprocess.new( + process = KaryaSpecSupport::E2ESubprocess.new( command_env, *framework_worker_command(framework:, queue:, worker_name:), chdir: current_app_root @@ -398,7 +400,7 @@ def runtime_control_rakefile_source(framework:) worker: true ) - process = E2ESubprocess.new( + process = KaryaSpecSupport::E2ESubprocess.new( worker_env, *framework_worker_command(framework:, queue:, worker_name:), chdir: current_app_root From 48d68fcd69a5e7d64a79d9b49024b2ebfb24355f Mon Sep 17 00:00:00 2001 From: Nitesh Purohit Date: Wed, 9 Sep 2026 19:42:46 -0400 Subject: [PATCH 4/5] refactor: Improve output timeout handling in E2ESubprocess --- spec/support/e2e_subprocess.rb | 3 ++- spec/support/framework_runtime_control_e2e_support.rb | 8 +++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb index 0e42fc5b..1abc26fc 100644 --- a/spec/support/e2e_subprocess.rb +++ b/spec/support/e2e_subprocess.rb @@ -11,6 +11,7 @@ module KaryaSpecSupport class E2ESubprocess class CleanupError < StandardError; end + class OutputTimeout < Timeout::Error; end DEFAULT_TIMEOUT = 30 TERMINATION_TIMEOUT = 2 @@ -74,7 +75,7 @@ def wait(timeout: DEFAULT_TIMEOUT) end def wait_for_output(timeout: TERMINATION_TIMEOUT) - Timeout.timeout(timeout) do + Timeout.timeout(timeout, OutputTimeout) do @stdout_reader.value @stderr_reader.value end diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb index b182ba9e..1c39b1b2 100644 --- a/spec/support/framework_runtime_control_e2e_support.rb +++ b/spec/support/framework_runtime_control_e2e_support.rb @@ -168,7 +168,11 @@ def wait_for_framework_runtime_phase(state_file, *phases) def wait_for_framework_runtime_start(state_file, process) wait_until do unless process.alive? - process.wait_for_output + begin + process.wait_for_output + rescue KaryaSpecSupport::E2ESubprocess::OutputTimeout + raise "worker exited before runtime control started and output remained open:\n#{process.output}" + end raise "worker exited before runtime control started:\n#{process.output}" end @@ -183,8 +187,6 @@ def wait_for_framework_runtime_start(state_file, process) payload rescue Errno::ENOENT, JSON::ParserError, KeyError next - rescue Timeout::Error - raise "worker exited before runtime control started and output remained open:\n#{process.output}" end end From f38067cefc14e999497b34ddaaccf2977e747673 Mon Sep 17 00:00:00 2001 From: Nitesh Purohit Date: Wed, 9 Sep 2026 19:53:33 -0400 Subject: [PATCH 5/5] refactor: Enhance subprocess cleanup handling --- gems/karya/spec/e2e/karya/cli_worker_spec.rb | 2 +- .../e2_e_subprocess_spec.rb | 36 +++++++++++++++++++ spec/support/e2e_subprocess.rb | 12 +++++-- .../framework_runtime_control_e2e_support.rb | 4 +-- 4 files changed, 49 insertions(+), 5 deletions(-) diff --git a/gems/karya/spec/e2e/karya/cli_worker_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_spec.rb index 2213888f..43014ff4 100644 --- a/gems/karya/spec/e2e/karya/cli_worker_spec.rb +++ b/gems/karya/spec/e2e/karya/cli_worker_spec.rb @@ -62,7 +62,7 @@ def with_force_stop_worker(boot_file:, state_file:, &) ), chdir: KaryaE2EHelpers::PACKAGE_ROOT) yield process ensure - process&.close + KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process) end def request_force_stop(supervisor_pid:, draining_marker_file:) diff --git a/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb index 5581d2f3..1b942214 100644 --- a/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb +++ b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb @@ -7,6 +7,7 @@ require_relative '../spec_helper' require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__) +require 'English' RSpec.describe KaryaSpecSupport::E2ESubprocess, :e2e, :integration do it 'bounds capture when a forked descendant retains an output pipe' do @@ -27,4 +28,39 @@ end end.to raise_error(Timeout::Error, /subprocess timed out after 0.5s/) end + + it 'terminates a descendant that outlives its leader without retaining output pipes' do + command = [ + RbConfig.ruby, + '-e', + <<~RUBY + fork do + 1.upto(255) do |file_descriptor| + IO.for_fd(file_descriptor).close + rescue Errno::EBADF + nil + end + Signal.trap('TERM', 'IGNORE') + loop { sleep 0.1 } + end + exit! 0 + RUBY + ] + + _stdout, _stderr, status = described_class.capture(*command, timeout: 2) + + expect(status).to be_success + end + + it 'preserves an active failure when cleanup also fails' do + process = instance_double(described_class) + cleanup_error = described_class::CleanupError.new('cleanup failed') + allow(process).to receive(:close).and_raise(cleanup_error) + + expect do + raise 'primary failure' + ensure + described_class.close_preserving_failure(process, active_exception: $ERROR_INFO) + end.to raise_error(RuntimeError, 'primary failure').and output(/cleanup failed/).to_stderr + end end diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb index 1abc26fc..e6165507 100644 --- a/spec/support/e2e_subprocess.rb +++ b/spec/support/e2e_subprocess.rb @@ -42,6 +42,14 @@ def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options) result end + def self.close_preserving_failure(process, active_exception: $!) + process&.close + rescue CleanupError => error + raise error unless active_exception + + warn "#{error.class}: #{error.message}" + end + def self.monotonic_time Process.clock_gettime(Process::CLOCK_MONOTONIC) end @@ -94,7 +102,7 @@ def output end def close(deadline: monotonic_time + TERMINATION_TIMEOUT) - terminate_process_group(deadline) if alive? || output_readers_alive? + terminate_process_group(deadline) unless shutdown_complete? close_output_streams join_until(@wait_thread, deadline) join_until(@stdout_reader, deadline) @@ -128,7 +136,7 @@ def terminate_process_group(deadline) signal_process_group('TERM') grace_deadline = monotonic_time + (remaining_time(deadline) / 2.0) join_until(@wait_thread, grace_deadline) - return unless alive? || output_readers_alive? + return if shutdown_complete? signal_process_group('KILL') end diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb index 1c39b1b2..4da22e58 100644 --- a/spec/support/framework_runtime_control_e2e_support.rb +++ b/spec/support/framework_runtime_control_e2e_support.rb @@ -378,7 +378,7 @@ def runtime_control_rakefile_source(framework:) state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)' raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" ensure - process.close + KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process) end end end @@ -446,7 +446,7 @@ def runtime_control_rakefile_source(framework:) raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}" ensure File.delete(stale_state_file) if File.exist?(stale_state_file) - process.close + KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process) end end end