Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion ruby/driver/riverqueue-activerecord/spec/spec_helper.rb
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,12 @@ def switch_to_postgres!
unless ENV["RIVERQUEUE_ROOT_TEST_SUITE"]
require "simplecov"
SimpleCov.start do
add_filter "/spec/"
skip "/spec/"
enable_coverage :branch
command_name RiverTestDatabase::BACKEND
minimum_coverage branch: 100, line: 100 if RiverTestDatabase::BACKEND == "all"
# A full run covers both backends without results cached by earlier runs.
merging false if RiverTestDatabase::BACKEND == "all"
end
end

Expand Down
4 changes: 3 additions & 1 deletion ruby/driver/riverqueue-sequel/spec/spec_helper.rb
Original file line number Diff line number Diff line change
Expand Up @@ -63,10 +63,12 @@ def available_test_transaction(&)
unless ENV["RIVERQUEUE_ROOT_TEST_SUITE"]
require "simplecov"
SimpleCov.start do
add_filter "/spec/"
skip "/spec/"
enable_coverage :branch
command_name RiverTestDatabase::BACKEND
minimum_coverage branch: 100, line: 100 if RiverTestDatabase::BACKEND == "all"
# A full run covers both backends without results cached by earlier runs.
merging false if RiverTestDatabase::BACKEND == "all"
end
end

Expand Down
2 changes: 1 addition & 1 deletion ruby/rails/riverqueue-rails/spec/configuration_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@
RSpec.describe River::Generators::InstallGenerator do
it "installs a Ruby configuration and executable worker command" do
Dir.mktmpdir("river-generator-") do |directory|
described_class.start([], destination_root: directory, shell: Thor::Shell::Basic.new)
described_class.start(["--quiet"], destination_root: directory, shell: Thor::Shell::Basic.new)

expect(File.read(File.join(directory, "config/initializers/river.rb"))).to include("queue_adapter = :river")
command = File.join(directory, "bin/jobs")
Expand Down
79 changes: 45 additions & 34 deletions ruby/spec/config_spec.rb
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# frozen_string_literal: true

require "spec_helper"
require "open3"

RSpec.describe River::QueueConfig do
%i[fetch_cooldown fetch_poll_interval].product([Float::NAN, Float::INFINITY, -Float::INFINITY]).each do |name, value|
Expand Down Expand Up @@ -152,25 +153,38 @@
end

describe "#id" do
# Runs the block in a forked child and returns its result, marshaled back
# through a pipe. The child exits with exit! to skip RSpec and coverage
# at_exit hooks.
def in_fork
reader, writer = IO.pipe
pid = Process.fork do
reader.close
writer.binmode.write(Marshal.dump(yield))
writer.close
exit!(0)
end
writer.close
result = Marshal.load(reader.binmode.read) # rubocop:disable Security/MarshalLoad
_, status = Process.wait2(pid)
raise "forked child failed: #{status.inspect}" unless status.success?
# Fork in a fresh Ruby process so unrelated specs' open SQLite connections
# aren't inherited by children that only exercise configuration IDs.
def ids_across_fork(count: 1, id: nil, read_parent_id: true)
output, error, status = Open3.capture3(RbConfig.ruby, "-Ilib", "-rriverqueue", "-e", <<~RUBY)
config = River::Config.new(id: #{id.inspect})
parent_id = config.id if #{read_parent_id}
children = Array.new(#{count}) do
reader, writer = IO.pipe
pid = Process.fork do
reader.close
client = River::Client.new(Object.new, config: config)
writer.binmode.write(Marshal.dump({
client_id: client.id, config_id: config.id,
copy_id: config.with(job_timeout: 5).id, pid: Process.pid
}))
writer.close
exit!(0)
end
writer.close
result = reader.binmode.read
reader.close
_, status = Process.wait2(pid)
raise "forked child failed: \#{status.inspect}" unless status.success?

Marshal.load(result)
end
$stdout.binmode.write(Marshal.dump({parent_id: parent_id, parent_id_after: config.id, children: children}))
RUBY

result
ensure
reader&.close
expect(status.success?).to be(true), error
expect(error).to eq("")
Marshal.load(output) # rubocop:disable Security/MarshalLoad
end

def skip_unless_fork_supported
Expand Down Expand Up @@ -200,31 +214,26 @@ def skip_unless_fork_supported
it "generates a new default ID in each forked process" do
skip_unless_fork_supported

config = described_class.new
parent_id = config.id
children = Array.new(2) do
in_fork do
client = River::Client.new(Object.new, config: config)
{client_id: client.id, config_id: config.id, copy_id: config.with(job_timeout: 5).id, pid: Process.pid}
end
end
result = ids_across_fork(count: 2)
parent_id = result.fetch(:parent_id)
children = result.fetch(:children)

children.each do |child|
expect(child.fetch(:client_id)).to eq(child.fetch(:config_id))
expect(child.fetch(:copy_id)).to eq(child.fetch(:config_id))
expect(child.fetch(:config_id)).to include("-#{child.fetch(:pid)}-")
end
expect(children.map { |child| child.fetch(:config_id) }.push(parent_id).uniq.length).to eq(3)
expect(config.id).to eq(parent_id)
expect(result.fetch(:parent_id_after)).to eq(parent_id)
end

it "generates a forked process's ID even when the parent never read one" do
skip_unless_fork_supported

config = described_class.new
child_id = in_fork { config.id }
result = ids_across_fork(read_parent_id: false)

expect(config.id).not_to eq(child_id)
expect(result.fetch(:parent_id)).to be_nil
expect(result.fetch(:parent_id_after)).not_to eq(result.fetch(:children).first.fetch(:config_id))
end

it "keeps a default ID stable within a process" do
Expand All @@ -247,11 +256,13 @@ def skip_unless_fork_supported
it "preserves an explicit ID across forks" do
skip_unless_fork_supported

config = described_class.new(id: "explicit-client")
child = in_fork { [config.id, River::Client.new(Object.new, config: config).id] }
result = ids_across_fork(id: "explicit-client")

expect(child).to eq(%w[explicit-client explicit-client])
expect(config.id).to eq("explicit-client")
expect(result.fetch(:children).first).to include(
client_id: "explicit-client", config_id: "explicit-client", copy_id: "explicit-client"
)
expect(result.fetch(:parent_id)).to eq("explicit-client")
expect(result.fetch(:parent_id_after)).to eq("explicit-client")
end

it "shares a default ID with copies unless overridden" do
Expand Down
8 changes: 6 additions & 2 deletions ruby/spec/notification_receiver_shared_examples.rb
Original file line number Diff line number Diff line change
Expand Up @@ -89,9 +89,13 @@ def receiver_client(**options, &work)
client.start
listener = receive_signal(listeners)
receive_signal(fetches)
pid = listener.instance_variable_get(:@connection).backend_pid
@driver.send(:runtime_execute, "SELECT pg_terminate_backend(#{pid})")
connection = listener.instance_variable_get(:@connection)
notices = Queue.new
# libpq otherwise prints the deliberately terminated connection's FATAL notice.
connection.set_notice_receiver { |result| notices << result.error_field(PG::PG_DIAG_SQLSTATE) }
@driver.send(:runtime_execute, "SELECT pg_terminate_backend(#{connection.backend_pid})")
expect(receive_signal(listeners)).not_to equal(listener)
expect(receive_signal(notices)).to eq("57P01") # admin_shutdown, independent of server locale

row = River::Client.new(@driver).insert(River::JobArgsHash.new(:notification, {}), queue: :notifications).job
expect(receive_signal(worked)).to eq(row.id)
Expand Down
8 changes: 5 additions & 3 deletions ruby/spec/spec_helper.rb
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,14 @@
enable_coverage :branch
command_name RiverTestDatabase::BACKEND
minimum_coverage branch: 100, line: 100 if RiverTestDatabase::BACKEND == "all"
# A full run covers both backends without results cached by earlier runs.
merging false if RiverTestDatabase::BACKEND == "all"

# Drivers have their own spec suite where they're covered 100.0%, but
# they're not fully covered from this top level test suite.
add_filter("driver/riverqueue-sequel/")
add_filter("driver/riverqueue-activerecord/")
add_filter("/spec/")
skip("driver/riverqueue-sequel/")
skip("driver/riverqueue-activerecord/")
skip("/spec/")
end
end

Expand Down
Loading