diff --git a/lib/mongo/operation/result.rb b/lib/mongo/operation/result.rb index 2add26d6e0..8b3d2fed95 100644 --- a/lib/mongo/operation/result.rb +++ b/lib/mongo/operation/result.rb @@ -162,15 +162,6 @@ def acknowledged? !!@replies end - # Whether the result contains cursor_id - # - # @return [ true, false ] If the result contains cursor_id. - # - # @api private - def has_cursor_id? - acknowledged? && replies.last.respond_to?(:cursor_id) - end - # Get the cursor id if the response is acknowledged. # # @note Cursor ids of 0 indicate there is no cursor on the server. @@ -183,7 +174,17 @@ def has_cursor_id? # @since 2.0.0 # @api private def cursor_id - acknowledged? ? replies.last.cursor_id : 0 + return 0 unless acknowledged? + + if replies.last.respond_to?(:cursor_id) + # Legacy OP_REPLY replies expose the cursor id directly. + replies.last.cursor_id + elsif first_document.key?(CURSOR) + # OP_MSG command replies carry the cursor id in the `cursor` subdocument. + first_document[CURSOR][CURSOR_ID] || 0 + else + 0 + end end # Get the namespace of the cursor. The method should be defined in diff --git a/lib/mongo/operation/shared/executable.rb b/lib/mongo/operation/shared/executable.rb index e57c79ee9c..034838783d 100644 --- a/lib/mongo/operation/shared/executable.rb +++ b/lib/mongo/operation/shared/executable.rb @@ -64,14 +64,6 @@ def do_execute(connection, context, options = {}) end end - if result.has_cursor_id? && - connection.description.load_balancer? - if result.cursor_id == 0 - connection.unpin - else - connection.pin - end - end process_result(result, connection) end end diff --git a/lib/mongo/tracing/open_telemetry/command_tracer.rb b/lib/mongo/tracing/open_telemetry/command_tracer.rb index 5af152addc..27585839de 100644 --- a/lib/mongo/tracing/open_telemetry/command_tracer.rb +++ b/lib/mongo/tracing/open_telemetry/command_tracer.rb @@ -206,9 +206,23 @@ def session_attributes(message) # @param _context [ OpenTelemetry::Context ] the context (unused). # @param span [ OpenTelemetry::Trace::Span ] the current span. def process_cursor_context(result, _cursor_id, _context, span) - return unless result.has_cursor_id? && result.cursor_id.positive? + cursor_id = normalize_cursor_id(result.cursor_id) + return unless cursor_id.positive? - span.set_attribute('db.mongodb.cursor_id', result.cursor_id) + span.set_attribute('db.mongodb.cursor_id', cursor_id) + end + + # Normalizes a cursor id to a plain Integer. + # + # OP_MSG replies deserialized in :bson mode expose the cursor id as a + # BSON::Int64, which does not implement Numeric#positive? and is not a + # valid OpenTelemetry attribute type. + # + # @param cursor_id [ Integer | BSON::Int64 ] the raw cursor id. + # + # @return [ Integer ] the cursor id as an Integer. + def normalize_cursor_id(cursor_id) + cursor_id.is_a?(BSON::Int64) ? cursor_id.value : cursor_id end # Records error status code if the command failed. diff --git a/spec/mongo/operation/result_spec.rb b/spec/mongo/operation/result_spec.rb index c868645cff..36ae275c6a 100644 --- a/spec/mongo/operation/result_spec.rb +++ b/spec/mongo/operation/result_spec.rb @@ -83,22 +83,24 @@ def get_result(client) expect(result.cursor_id).to eq(0) end end - end - describe '#has_cursor_id?' do - context 'when the reply exists' do - let(:cursor_id) { 5 } + context 'when the reply is an OP_MSG carrying a cursor subdocument' do + let(:reply) do + Mongo::Protocol::Msg.new([], {}, { 'cursor' => { 'id' => 42 } }) + end - it 'returns true' do - expect(result).to have_cursor_id + it 'reads the cursor id from the cursor subdocument' do + expect(result.cursor_id).to eq(42) end end - context 'when the reply does not exist' do - let(:reply) { nil } + context 'when the reply is an OP_MSG without a cursor subdocument' do + let(:reply) do + Mongo::Protocol::Msg.new([], {}, { 'ok' => 1 }) + end - it 'returns false' do - expect(result).not_to have_cursor_id + it 'returns zero' do + expect(result.cursor_id).to eq(0) end end end diff --git a/spec/mongo/tracing/open_telemetry/command_tracer_spec.rb b/spec/mongo/tracing/open_telemetry/command_tracer_spec.rb index a386229aaa..28ac8b36fb 100644 --- a/spec/mongo/tracing/open_telemetry/command_tracer_spec.rb +++ b/spec/mongo/tracing/open_telemetry/command_tracer_spec.rb @@ -67,7 +67,7 @@ describe '#trace_command' do let(:span) { instance_double(OpenTelemetry::Trace::Span, finish: nil, set_attribute: nil) } let(:context) { instance_double(Mongo::Operation::Context) } - let(:result) { instance_double(Mongo::Operation::Result, has_cursor_id?: false, successful?: true) } + let(:result) { instance_double(Mongo::Operation::Result, cursor_id: 0, successful?: true) } before do allow(otel_tracer).to receive(:start_span).and_return(span) @@ -103,7 +103,7 @@ context 'when result has cursor_id' do let(:result) do - instance_double(Mongo::Operation::Result, has_cursor_id?: true, cursor_id: 789, successful?: true) + instance_double(Mongo::Operation::Result, cursor_id: 789, successful?: true) end it 'sets the cursor_id attribute' do @@ -112,9 +112,20 @@ end end + context 'when result cursor_id is a BSON::Int64' do + let(:result) do + instance_double(Mongo::Operation::Result, cursor_id: BSON::Int64.new(789), successful?: true) + end + + it 'records the cursor id as an Integer' do + expect(span).to receive(:set_attribute).with('db.mongodb.cursor_id', 789) + command_tracer.trace_command(message, operation_context, connection) { result } + end + end + context 'when result has zero cursor_id' do let(:result) do - instance_double(Mongo::Operation::Result, has_cursor_id?: true, cursor_id: 0, successful?: true) + instance_double(Mongo::Operation::Result, cursor_id: 0, successful?: true) end it 'does not set the cursor_id attribute' do