From 0e1b0a5d1273778d9bb7aa5aa959ab780642b032 Mon Sep 17 00:00:00 2001 From: annbelova1 Date: Sun, 5 Jul 2026 09:01:59 +0300 Subject: [PATCH 1/4] Refactor async_run to properly handle OpenTelemetry context propagation in async jobs --- lib/sequel/extensions/concurrent_thread_pool.rb | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/lib/sequel/extensions/concurrent_thread_pool.rb b/lib/sequel/extensions/concurrent_thread_pool.rb index cec619c..3802e74 100644 --- a/lib/sequel/extensions/concurrent_thread_pool.rb +++ b/lib/sequel/extensions/concurrent_thread_pool.rb @@ -120,14 +120,14 @@ def choose_executor(opts) end def async_run(&) - if defined?(OpenTelemetry) + if defined?(OpenTelemetry::Context) otel_context = OpenTelemetry::Context.current - async_job_class.new(async_thread_executor) do - token = OpenTelemetry::Context.attach(otel_context) - begin - yield - ensure - OpenTelemetry::Context.detach(token) + + if otel_context.equal?(OpenTelemetry::Context::ROOT) + async_job_class.new(async_thread_executor, &) + else + async_job_class.new(async_thread_executor) do + OpenTelemetry::Context.with_current(otel_context, &) end end else From 2ace37e82bb6b7527829049a8d22c8135ac3f1c6 Mon Sep 17 00:00:00 2001 From: annbelova1 Date: Sun, 5 Jul 2026 09:12:47 +0300 Subject: [PATCH 2/4] Refactor rspec by isolating OpenTelemetry context propagation logic --- .../concurrent_thread_pool_helpers.rb | 9 +++ ...ncurrent_thread_pool_opentelemetry_spec.rb | 68 +++++++++++++++++++ .../extensions/concurrent_thread_pool_spec.rb | 63 +---------------- 3 files changed, 78 insertions(+), 62 deletions(-) create mode 100644 spec/extensions/concurrent_thread_pool_helpers.rb create mode 100644 spec/extensions/concurrent_thread_pool_opentelemetry_spec.rb diff --git a/spec/extensions/concurrent_thread_pool_helpers.rb b/spec/extensions/concurrent_thread_pool_helpers.rb new file mode 100644 index 0000000..e41e9f3 --- /dev/null +++ b/spec/extensions/concurrent_thread_pool_helpers.rb @@ -0,0 +1,9 @@ +# frozen_string_literal: true + +require "concurrent" + +ASYNC_DB_URL = ENV.fetch("DB_URL", "postgres:///sequel_plugins") + +def make_concurrent_db(**opts) + Sequel.connect(ASYNC_DB_URL, **opts).tap { |d| d.extension(:concurrent_thread_pool) } +end diff --git a/spec/extensions/concurrent_thread_pool_opentelemetry_spec.rb b/spec/extensions/concurrent_thread_pool_opentelemetry_spec.rb new file mode 100644 index 0000000..c8532cd --- /dev/null +++ b/spec/extensions/concurrent_thread_pool_opentelemetry_spec.rb @@ -0,0 +1,68 @@ +# frozen_string_literal: true + +require "opentelemetry-api" +require_relative "concurrent_thread_pool_helpers" + +RSpec.describe "concurrent_thread_pool extension OpenTelemetry context propagation" do + let(:db) { make_concurrent_db } + let(:otel_key) { OpenTelemetry::Context.create_key("sequel-test") } + + after { db.disconnect rescue nil } + + it "does not propagate context when calling thread context is ROOT" do + captured = :not_set + + db.send(:async_run) { captured = OpenTelemetry::Context.current.value(otel_key) }.__value + + expect(captured).to be_nil + end + + it "does not error when OpenTelemetry is defined without Context" do + db = make_concurrent_db + stub_const("OpenTelemetry", Object.new) + + expect { db.send(:async_run) { :ok }.__value }.not_to raise_error + ensure + db.disconnect rescue nil + end + + it "propagates calling-thread context to worker thread" do + captured = nil + + OpenTelemetry::Context.with_value(otel_key, "sentinel") do + db.send(:async_run) { captured = OpenTelemetry::Context.current.value(otel_key) }.__value + end + + expect(captured).to eq("sentinel") + end + + context "with single-thread executor" do + let(:db) { make_concurrent_db(num_async_threads: 1) } + + it "restores context in worker thread after block completes" do + OpenTelemetry::Context.with_value(otel_key, "sentinel") do + db.send(:async_run) { :done }.__value + + value_after = Concurrent::Promises.future_on(db.async_thread_executor) do + OpenTelemetry::Context.current.value(otel_key) + end.value! + + expect(value_after).to be_nil + end + end + end + + it "propagates context even when block raises" do + captured = nil + + OpenTelemetry::Context.with_value(otel_key, "sentinel") do + result = db.send(:async_run) do + captured = OpenTelemetry::Context.current.value(otel_key) + raise "boom" + end + + expect { result.__value }.to raise_error(RuntimeError, "boom") + expect(captured).to eq("sentinel") + end + end +end diff --git a/spec/extensions/concurrent_thread_pool_spec.rb b/spec/extensions/concurrent_thread_pool_spec.rb index b69e257..b4878d6 100644 --- a/spec/extensions/concurrent_thread_pool_spec.rb +++ b/spec/extensions/concurrent_thread_pool_spec.rb @@ -1,12 +1,6 @@ # frozen_string_literal: true -require "concurrent" - -ASYNC_DB_URL = ENV.fetch("DB_URL", "postgres:///sequel_plugins") - -def make_concurrent_db(**opts) - Sequel.connect(ASYNC_DB_URL, **opts).tap { |d| d.extension(:concurrent_thread_pool) } -end +require_relative "concurrent_thread_pool_helpers" RSpec.describe "concurrent_thread_pool extension" do let(:db) { make_concurrent_db(**db_opts) } @@ -172,61 +166,6 @@ def make_concurrent_db(**opts) end end - describe "OpenTelemetry context propagation" do - require "opentelemetry-api" - - let(:otel_key) { OpenTelemetry::Context.create_key("sequel-test") } - - after do - # ensure no context leak between examples - OpenTelemetry::Context.detach(OpenTelemetry::Context.attach(OpenTelemetry::Context::ROOT)) - end - - it "propagates calling-thread context to worker thread" do - ctx = OpenTelemetry::Context.current.set_value(otel_key, "sentinel") - token = OpenTelemetry::Context.attach(ctx) - captured = nil - - db.send(:async_run) { captured = OpenTelemetry::Context.current.value(otel_key) }.__value - - expect(captured).to eq("sentinel") - ensure - OpenTelemetry::Context.detach(token) - end - - it "restores context in worker thread after block completes" do - ctx = OpenTelemetry::Context.current.set_value(otel_key, "sentinel") - token = OpenTelemetry::Context.attach(ctx) - value_after = :not_set - - db.send(:async_run) do - inner_token = OpenTelemetry::Context.attach(OpenTelemetry::Context::ROOT) - OpenTelemetry::Context.detach(inner_token) - value_after = OpenTelemetry::Context.current.value(otel_key) - end.__value - - expect(value_after).to eq("sentinel") - ensure - OpenTelemetry::Context.detach(token) - end - - it "propagates context even when block raises" do - ctx = OpenTelemetry::Context.current.set_value(otel_key, "sentinel") - token = OpenTelemetry::Context.attach(ctx) - captured = nil - - result = db.send(:async_run) do - captured = OpenTelemetry::Context.current.value(otel_key) - raise "boom" - end - - expect { result.__value }.to raise_error(RuntimeError, "boom") - expect(captured).to eq("sentinel") - ensure - OpenTelemetry::Context.detach(token) - end - end - describe "async_job_class" do it "is Proxy by default" do expect(db.async_job_class).to eq(Sequel::Database::ConcurrentThreadPool::Proxy) From 842c6bfedc4f3ede3169a2342bc3079e1f497a51 Mon Sep 17 00:00:00 2001 From: annbelova1 Date: Sun, 5 Jul 2026 09:22:09 +0300 Subject: [PATCH 3/4] Update upsert method documentation to clarify parameters and usage example --- lib/sequel/plugins/upsert.rb | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/sequel/plugins/upsert.rb b/lib/sequel/plugins/upsert.rb index 07fe887..f6f4e0a 100644 --- a/lib/sequel/plugins/upsert.rb +++ b/lib/sequel/plugins/upsert.rb @@ -25,10 +25,10 @@ def upsert_dataset(target: primary_key) # Executes the upsert request # # @param row [Hash] values - # @param options [Hash] options + # @param target [Symbol] target column # # @example - # User.upsert(name: "John", email: "jd@test.com", target: :email) + # User.upsert({name: "John", email: "jd@test.com"}, target: :email) # @return [Sequel::Model] def upsert(row, **) upsert_dataset(**).insert(sequel_values(row)) From 1f9504f354934d2b82d434eb86b3cf1f04997e2c Mon Sep 17 00:00:00 2001 From: annbelova1 Date: Sun, 5 Jul 2026 09:24:52 +0300 Subject: [PATCH 4/4] Bump umbrellio-sequel-plugins version to 0.19.0 --- Gemfile.lock | 2 +- umbrellio-sequel-plugins.gemspec | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Gemfile.lock b/Gemfile.lock index 5c88127..2137fe7 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - umbrellio-sequel-plugins (0.18.1) + umbrellio-sequel-plugins (0.19.0) concurrent-ruby sequel diff --git a/umbrellio-sequel-plugins.gemspec b/umbrellio-sequel-plugins.gemspec index 7eac906..0511212 100644 --- a/umbrellio-sequel-plugins.gemspec +++ b/umbrellio-sequel-plugins.gemspec @@ -5,7 +5,7 @@ $LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib) Gem::Specification.new do |spec| spec.name = "umbrellio-sequel-plugins" - spec.version = "0.18.1" + spec.version = "0.19.0" spec.required_ruby_version = ">= 3.3" spec.authors = ["Team Umbrellio"]