28930904828

Committed 08 Jul 2026 09:05AM UTC coverage: 97.391% (+0.01%) from 97.378%

Build # 28930904828

Build Type

push

github

Committed by

web-flow

Commit Message

Propagate OpenTelemetry context to async thread pool workers (#46)

* propagate OpenTelemetry context to async thread pool workers

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix rubocop offenses

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* use real opentelemetry-api gem in tests instead of stubs

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* drop Ruby < 3.3 support, add 3.4 to CI matrix

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* bump TargetRubyVersion to 3.3, apply new style cops

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Refactor OpenTelemetry context propagation and cleanup specs (#47)

* Refactor async_run to properly handle OpenTelemetry context propagation in async jobs

* Refactor rspec by isolating OpenTelemetry context propagation logic

* Update upsert method documentation to clarify parameters and usage example

* Bump umbrellio-sequel-plugins version to 0.19.0

* add release workflow

* fix release workflow

* style fix

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: avoleba <52998202+avoleba@users.noreply.github.com>

Coverage Stats

21 of 22 new or added lines in 5 files covered. (95.45%)

560 of 575 relevant lines covered (97.39%)

9.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.88

/lib/sequel/extensions/concurrent_thread_pool.rb

# frozen_string_literal: true

require "concurrent"

module Sequel
  # https://github.com/jeremyevans/sequel/blob/master/lib/sequel/extensions/async_thread_pool.rb
  module Database::ConcurrentThreadPool
    # Base proxy: delegates all method calls to the resolved async value.
    class BaseProxy < BasicObject
      def method_missing(...)
        __value.public_send(...)
      end

      def respond_to_missing?(*)
        __value.respond_to?(*)
      end

      [:!, :==, :!=, :instance_eval, :instance_exec].each do |method|
        define_method(method) do |*args, &block|
          __value.public_send(method, *args, &block)
        end
      end
    end

    # Default proxy: schedules block via Concurrent::Future, blocks on first access.
    class Proxy < BaseProxy
      def initialize(executor, &)
        super()

        @future = Concurrent::Promises.future_on(executor, &)
      end

      def __value
        @future.value!
      end
    end

    # Preemptable proxy: calling thread runs the block if the pool hasn't started it yet.
    class PreemptableProxy < BaseProxy
      def initialize(executor, &block)
        super()

        @mutex = Mutex.new
        @block = block
        @done = false
        @result = nil
        @error = nil

        executor.post { __run }
      end

      def __value
        error, result = @mutex.synchronize do
          __execute unless @done
          [@error, @result]
        end
        ::Kernel.raise error if error
        result
      end

      private

      def __run
        @mutex.synchronize { __execute unless @done }
      end

      def __execute
        @result = @block.call
      rescue StandardError => error
        @error = error
      ensure
        @done = true
      end
    end

    module DatabaseMethods
      class << self
        def make_finalizer(executor) = proc { executor.shutdown }

        def extended(db)
          db.instance_exec do
            case pool.pool_type
            when :single, :sharded_single
              raise Error, "cannot load concurrent_thread_pool extension " \
                           "if using single or sharded_single connection pool"
            end

            executor, owned = choose_executor(opts)
            proxy_klass =
              typecast_value_boolean(opts[:preempt_async_thread]) ? PreemptableProxy : Proxy

            define_singleton_method(:async_job_class) { proxy_klass }
            define_singleton_method(:async_thread_executor) { executor }

            finalizer =
              Sequel::Database::ConcurrentThreadPool::DatabaseMethods.make_finalizer(executor)
            ObjectSpace.define_finalizer(db, finalizer) if owned

            extend_datasets(DatasetMethods)
          end
        end
      end

      private

      def choose_executor(opts)
        if opts[:async_thread_executor]
          [opts[:async_thread_executor], false]
        else
          num = opts[:num_async_threads] ? typecast_value_integer(opts[:num_async_threads]) :
                                Integer(opts[:max_connections] || 4)
          raise Error, "must have positive number for num_async_threads" if num <= 0
          [Concurrent::ThreadPoolExecutor.new(
            min_threads: num,
            max_threads: num,
            max_queue: 0,
            fallback_policy: :abort,
          ), true]
        end
      end

      def async_run(&)
        otel_context = OpenTelemetry::Context.current if defined?(OpenTelemetry::Context)

        if otel_context && !otel_context.equal?(OpenTelemetry::Context::ROOT)
          async_job_class.new(async_thread_executor) do
            OpenTelemetry::Context.with_current(otel_context, &)
          end
        else
          async_job_class.new(async_thread_executor, &)
        end
      end
    end

    ASYNC_METHODS = ((
      [:all?, :any?, :drop, :entries, :grep_v, :include?, :inject, :member?, :minmax,
       :none?, :one?, :reduce, :sort, :take, :tally, :to_a, :to_h, :uniq, :zip] &
        Enumerable.instance_methods
    ) + (Dataset::ACTION_METHODS - [:map, :paged_each])).freeze

    ASYNC_BLOCK_METHODS = ((
      [:collect, :collect_concat, :detect, :drop_while, :each_cons, :each_entry, :each_slice,
       :each_with_index, :each_with_object, :filter_map, :find, :find_all, :find_index,
       :flat_map, :max_by, :min_by, :minmax_by, :partition, :reject, :reverse_each,
       :sort_by, :take_while] & Enumerable.instance_methods
    ) + [:paged_each]).freeze

    ASYNC_ARGS_OR_BLOCK_METHODS = [:map].freeze

    module DatasetMethods
      def self.define_async_method(mod, method)
        mod.send(:define_method, method) do |*args, &block|
          if @opts[:async]
            ds = sync
            db.send(:async_run) { ds.send(method, *args, &block) }
          else
            super(*args, &block)
          end
        end
      end

      def self.define_async_block_method(mod, method)
        mod.send(:define_method, method) do |*args, &block|
          if block && @opts[:async]
            ds = sync
            db.send(:async_run) { ds.send(method, *args, &block) }
          else
            super(*args, &block)
          end
        end
      end

      def self.define_async_args_or_block_method(mod, method)
        mod.send(:define_method, method) do |*args, &block|
          if (block || !args.empty?) && @opts[:async]
            ds = sync
            db.send(:async_run) { ds.send(method, *args, &block) }
          else
            super(*args, &block)
          end
        end
      end

      ASYNC_METHODS.each { |m| define_async_method(self, m) }
      ASYNC_BLOCK_METHODS.each { |m| define_async_block_method(self, m) }
      ASYNC_ARGS_OR_BLOCK_METHODS.each { |m| define_async_args_or_block_method(self, m) }

      def async
        cached_dataset(:_async) { clone(async: true) }
      end

      def sync
        cached_dataset(:_sync) { clone(async: false) }
      end
    end
  end

  Database.register_extension(
    :concurrent_thread_pool, Database::ConcurrentThreadPool::DatabaseMethods
  )
end

1	# frozen_string_literal: true
2
3	require "concurrent"	2✔
4
5	module Sequel	2✔
6	# https://github.com/jeremyevans/sequel/blob/master/lib/sequel/extensions/async_thread_pool.rb
7	module Database::ConcurrentThreadPool	2✔
8	# Base proxy: delegates all method calls to the resolved async value.
9	class BaseProxy < BasicObject	2✔
10	def method_missing(...)	2✔
11	__value.public_send(...)	20✔
12	end
13
14	def respond_to_missing?(*)	2✔
NEW 15	__value.respond_to?(*)	×
16	end
17
18	[:!, :==, :!=, :instance_eval, :instance_exec].each do \|method\|	2✔
19	define_method(method) do \|*args, &block\|	10✔
20	__value.public_send(method, *args, &block)	×
21	end
22	end
23	end
24
25	# Default proxy: schedules block via Concurrent::Future, blocks on first access.
26	class Proxy < BaseProxy	2✔
27	def initialize(executor, &)	2✔
28	super()	36✔
29
30	@future = Concurrent::Promises.future_on(executor, &)	36✔
31	end
32
33	def __value	2✔
34	@future.value!	54✔
35	end
36	end
37
38	# Preemptable proxy: calling thread runs the block if the pool hasn't started it yet.
39	class PreemptableProxy < BaseProxy	2✔
40	def initialize(executor, &block)	2✔
41	super()	6✔
42
43	@mutex = Mutex.new	6✔
44	@block = block	6✔
45	@done = false	6✔
46	@result = nil	6✔
47	@error = nil	6✔
48
49	executor.post { __run }	12✔
50	end
51
52	def __value	2✔
53	error, result = @mutex.synchronize do	20✔
54	__execute unless @done	20✔
55	[@error, @result]	20✔
56	end
57	::Kernel.raise error if error	20✔
58	result	18✔
59	end
60
61	private	2✔
62
63	def __run	2✔
64	@mutex.synchronize { __execute unless @done }	12✔
65	end
66
67	def __execute	2✔
68	@result = @block.call	6✔
69	rescue StandardError => error
70	@error = error	2✔
71	ensure
72	@done = true	6✔
73	end
74	end
75
76	module DatabaseMethods	2✔
77	class << self	2✔
78	def make_finalizer(executor) = proc { executor.shutdown }	2✔
79
80	def extended(db)	2✔
81	db.instance_exec do	118✔
82	case pool.pool_type	118✔
83	when :single, :sharded_single
84	raise Error, "cannot load concurrent_thread_pool extension " \	2✔
85	"if using single or sharded_single connection pool"
86	end
87
88	executor, owned = choose_executor(opts)	116✔
89	proxy_klass =
90	typecast_value_boolean(opts[:preempt_async_thread]) ? PreemptableProxy : Proxy	114✔
91
92	define_singleton_method(:async_job_class) { proxy_klass }	160✔
93	define_singleton_method(:async_thread_executor) { executor }	168✔
94
95	finalizer =
96	Sequel::Database::ConcurrentThreadPool::DatabaseMethods.make_finalizer(executor)	114✔
97	ObjectSpace.define_finalizer(db, finalizer) if owned	114✔
98
99	extend_datasets(DatasetMethods)	114✔
100	end
101	end
102	end
103
104	private	2✔
105
106	def choose_executor(opts)	2✔
107	if opts[:async_thread_executor]	116✔
108	[opts[:async_thread_executor], false]	2✔
109	else
110	num = opts[:num_async_threads] ? typecast_value_integer(opts[:num_async_threads]) :	114✔
111	Integer(opts[:max_connections] \|\| 4)	53✔
112	raise Error, "must have positive number for num_async_threads" if num <= 0	114✔
113	[Concurrent::ThreadPoolExecutor.new(	112✔
114	min_threads: num,
115	max_threads: num,
116	max_queue: 0,
117	fallback_policy: :abort,
118	), true]
119	end
120	end
121
122	def async_run(&)	2✔
123	otel_context = OpenTelemetry::Context.current if defined?(OpenTelemetry::Context)	42✔
124
125	if otel_context && !otel_context.equal?(OpenTelemetry::Context::ROOT)	42✔
126	async_job_class.new(async_thread_executor) do	6✔
127	OpenTelemetry::Context.with_current(otel_context, &)	6✔
128	end
129	else
130	async_job_class.new(async_thread_executor, &)	36✔
131	end
132	end
133	end
134
135	ASYNC_METHODS = ((
136	[:all?, :any?, :drop, :entries, :grep_v, :include?, :inject, :member?, :minmax,	2✔
137	:none?, :one?, :reduce, :sort, :take, :tally, :to_a, :to_h, :uniq, :zip] &
138	Enumerable.instance_methods
139	) + (Dataset::ACTION_METHODS - [:map, :paged_each])).freeze	2✔
140
141	ASYNC_BLOCK_METHODS = ((
142	[:collect, :collect_concat, :detect, :drop_while, :each_cons, :each_entry, :each_slice,	2✔
143	:each_with_index, :each_with_object, :filter_map, :find, :find_all, :find_index,
144	:flat_map, :max_by, :min_by, :minmax_by, :partition, :reject, :reverse_each,
145	:sort_by, :take_while] & Enumerable.instance_methods
146	) + [:paged_each]).freeze
147
148	ASYNC_ARGS_OR_BLOCK_METHODS = [:map].freeze	2✔
149
150	module DatasetMethods	2✔
151	def self.define_async_method(mod, method)	2✔
152	mod.send(:define_method, method) do \|*args, &block\|	112✔
153	if @opts[:async]	790✔
154	ds = sync	30✔
155	db.send(:async_run) { ds.send(method, *args, &block) }	60✔
156	else
157	super(*args, &block)	760✔
158	end
159	end
160	end
161
162	def self.define_async_block_method(mod, method)	2✔
163	mod.send(:define_method, method) do \|*args, &block\|	46✔
164	if block && @opts[:async]	×
165	ds = sync	×
166	db.send(:async_run) { ds.send(method, *args, &block) }	×
167	else
168	super(*args, &block)	×
169	end
170	end
171	end
172
173	def self.define_async_args_or_block_method(mod, method)	2✔
174	mod.send(:define_method, method) do \|*args, &block\|	2✔
175	if (block \|\| !args.empty?) && @opts[:async]	4✔
176	ds = sync	2✔
177	db.send(:async_run) { ds.send(method, *args, &block) }	4✔
178	else
179	super(*args, &block)	2✔
180	end
181	end
182	end
183
184	ASYNC_METHODS.each { \|m\| define_async_method(self, m) }	114✔
185	ASYNC_BLOCK_METHODS.each { \|m\| define_async_block_method(self, m) }	48✔
186	ASYNC_ARGS_OR_BLOCK_METHODS.each { \|m\| define_async_args_or_block_method(self, m) }	4✔
187
188	def async	2✔
189	cached_dataset(:_async) { clone(async: true) }	72✔
190	end
191
192	def sync	2✔
193	cached_dataset(:_sync) { clone(async: false) }	72✔
194	end
195	end
196	end
197
198	Database.register_extension(	2✔
199	:concurrent_thread_pool, Database::ConcurrentThreadPool::DatabaseMethods
200	)
201	end

umbrellio / umbrellio-sequel-plugins / 28930904828

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous