feat(queues): improve core queue conformance

Author: solnic

sentry-ruby/lib/sentry-ruby.rb

@@ -12,6 +12,7 @@
 require "sentry/utils/logging_helper"
 require "sentry/utils/callback_helper"
 require "sentry/utils/sample_rand"
+require "sentry/utils/latency"
 require "sentry/configuration"
 require "sentry/structured_logger"
 require "sentry/debug_structured_logger"

sentry-ruby/lib/sentry/job_extensions/extension.rb

@@ -181,7 +181,9 @@ def extract_trace_headers(job:, payload:)
       # @return [Hash] a Hash with any of the keys +:system+ (a String naming
       #   the backend), +:message_id+ (a String), +:destination+ (a String,
       #   such as the queue name), +:receive_latency+ (a Numeric, the
-      #   milliseconds between enqueueing and the start of the job). The fallback result is an empty Hash.
+      #   milliseconds between enqueueing and the start of the job) and
+      #   +:body_size+ (a non-negative Integer, the size of the message body
+      #   in bytes). The fallback result is an empty Hash.
       def messaging_data(job:)
         {}
       end

sentry-ruby/lib/sentry/job_extensions/instrumentation.rb

@@ -8,7 +8,8 @@ class Instrumentation
         system: Span::DataConventions::MESSAGING_SYSTEM,
         message_id: Span::DataConventions::MESSAGING_MESSAGE_ID,
         destination: Span::DataConventions::MESSAGING_DESTINATION_NAME,
-        receive_latency: Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY
+        receive_latency: Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY,
+        body_size: Span::DataConventions::MESSAGING_MESSAGE_BODY_SIZE
       }.freeze
 
       attr_reader :registry

sentry-ruby/lib/sentry/job_extensions/validators.rb

@@ -14,7 +14,8 @@ class Validators
         system: ->(value) { value.is_a?(String) },
         message_id: ->(value) { value.is_a?(String) },
         destination: ->(value) { value.is_a?(String) },
-        receive_latency: ->(value) { value.is_a?(Numeric) }
+        receive_latency: ->(value) { value.is_a?(Numeric) },
+        body_size: ->(value) { value.is_a?(Integer) && value >= 0 }
       }.freeze
 
       def initialize(event_keys: EVENT_DATA_KEYS, scope_keys: SCOPE_DATA_KEYS, messaging_fields: MESSAGING_DATA_VALIDATORS)

sentry-ruby/lib/sentry/span.rb

@@ -50,6 +50,7 @@ module DataConventions
       MESSAGING_DESTINATION_NAME = "messaging.destination.name"
       MESSAGING_MESSAGE_RECEIVE_LATENCY = "messaging.message.receive.latency"
       MESSAGING_MESSAGE_RETRY_COUNT = "messaging.message.retry.count"
+      MESSAGING_MESSAGE_BODY_SIZE = "messaging.message.body.size"
 
       # Time in ms the request spent in the server queue before processing began.
       HTTP_QUEUE_TIME_MS = "http.server.request.time_in_queue"

sentry-ruby/lib/sentry/utils/latency.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+require "time"
+
+module Sentry
+  module Utils
+    module Latency
+      module_function
+
+      def milliseconds_since(time)
+        return unless time
+
+        time = Time.parse(time) if time.is_a?(String)
+        (Time.now.to_f - time.to_f) * 1000.0
+      end
+    end
+  end
+end

sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb

@@ -48,7 +48,7 @@ def capture(consumer, exception = RuntimeError.new("boom"), **options)
 
   def enriching_consumer
     consumer(
-      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5, body_size: 128 } },
       retry_info: ->(job:) { { count: 3, will_retry: true } },
       scope_data: ->(job:) { { contexts: { backend: { "jid" => "backend-jid" } }, tags: { "backend.queue" => "low" }, extra: { size: 12 } } }
     )
@@ -112,6 +112,7 @@ def enriching_consumer
         "messaging.message.id" => "job-1",
         "messaging.destination.name" => "critical",
         "messaging.message.receive.latency" => 42.5,
+        "messaging.message.body.size" => 128,
         "messaging.message.retry.count" => 3
       )
     end

sentry-ruby/spec/sentry/job_extensions/producer_spec.rb

@@ -18,7 +18,7 @@ def producer(origin: nil, **hooks)
   def enriching_producer(origin: nil)
     producer(
       origin: origin,
-      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5, body_size: 128 } },
       retry_info: ->(job:) { { count: 3, will_retry: true } }
     )
   end
@@ -62,6 +62,7 @@ def record_span(producer, **options, &enqueue)
         "messaging.message.id" => "job-1",
         "messaging.destination.name" => "critical",
         "messaging.message.receive.latency" => 42.5,
+        "messaging.message.body.size" => 128,
         "messaging.message.retry.count" => 3
       )
     end

sentry-ruby/spec/sentry/job_extensions_spec.rb

@@ -418,12 +418,12 @@ def messaging_data(job:)
     it "returns the messaging span data an extension derives from the job" do
       extension = build_hooked_extension(
         messaging_data: lambda do |job:|
-          { system: "sidekiq", message_id: job.jid, destination: job.queue, receive_latency: 1500 }
+          { system: "sidekiq", message_id: job.jid, destination: job.queue, receive_latency: 1500, body_size: 42 }
         end
       )
 
       expect(call_hook(extension, :messaging_data))
-        .to eq(system: "sidekiq", message_id: "abc123", destination: "default", receive_latency: 1500)
+        .to eq(system: "sidekiq", message_id: "abc123", destination: "default", receive_latency: 1500, body_size: 42)
     end
 
     it "returns the names an extension gives the job and the integration it captures errors under" do
@@ -525,7 +525,7 @@ def messaging_data(job:)
       inject_trace_headers: [nil, "payload"],
       extract_trace_headers: ["sentry-trace", [], { "sentry-trace" => 1 }, { sentry_trace: "trace" }],
       messaging_data: [nil, [], { system: :sidekiq }, { message_id: 1 }, { destination: nil },
-                       { receive_latency: "fast" }],
+                       { receive_latency: "fast" }, { body_size: -1 }, { body_size: 1.5 }, { body_size: "42" }],
       transaction_name: ["", :HardJob, 1],
       span_origin: ["", :"auto.queue.backend"],
       publish_description: ["", :HardJob],

sentry-ruby/spec/sentry/utils/latency_spec.rb

@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+RSpec.describe Sentry::Utils::Latency do
+  let(:started_at) { Time.utc(2026, 1, 1, 12, 0, 0) }
+
+  around do |example|
+    Timecop.freeze(started_at + 2.5) { example.run }
+  end
+
+  it "measures the milliseconds since a Time" do
+    expect(described_class.milliseconds_since(started_at)).to eq(2_500.0)
+  end
+
+  it "measures the milliseconds since an ISO 8601 String" do
+    expect(described_class.milliseconds_since(started_at.iso8601(3))).to eq(2_500.0)
+  end
+
+  it "measures nothing without a time" do
+    expect(described_class.milliseconds_since(nil)).to be_nil
+  end
+end