Use record created_at as enqueued_at timestamp when pushing a job to Sidekiq

This allows Sidekiq to calculate the latency including the time a job has spent in outbox

Author: jakubkosinski

contrib/ruby_event_store-outbox/CHANGELOG.md

@@ -1,43 +1,47 @@
-### 0.1.0  2025-03-25
+### unreleased
 
-* Add support for non-locking repository that is using `SKIP LOCKED` clause and doesn't use global locks.
+- change the timestamp reported in `enqueued_at` to Sidekiq when pushing jobs from the current time to the record created at timestamp (so that the latency reported in Sidekiq includes the time a job has spent in outbox)
 
-### 0.0.30  2025-03-25
+### 0.1.0 2025-03-25
 
-* Fix an issue with `res_outbox` not starting due to inability to require `sidekiq`.
+- Add support for non-locking repository that is using `SKIP LOCKED` clause and doesn't use global locks.
+
+### 0.0.30 2025-03-25
+
+- Fix an issue with `res_outbox` not starting due to inability to require `sidekiq`.
   Sidekiq is only available in a producer context, while res_outbox is a consumer.
 
-### 0.0.29  2025-03-25
+### 0.0.29 2025-03-25
 
-* Fix an issue with uninitialized constant `RubyEventStore::Outbox::RetriableError` when using `bin/res_outbox`
+- Fix an issue with uninitialized constant `RubyEventStore::Outbox::RetriableError` when using `bin/res_outbox`
 
-### 0.0.28  2024-04-12
+### 0.0.28 2024-04-12
 
-* Fix issues that prevent res_outbox CLI from processing
-* Upgrade docker image base to ruby:3.2
+- Fix issues that prevent res_outbox CLI from processing
+- Upgrade docker image base to ruby:3.2
 
-### 0.0.27  2024-04-12
+### 0.0.27 2024-04-12
 
-* Fix issues that prevent res_outbox CLI from starting
+- Fix issues that prevent res_outbox CLI from starting
 
-### 0.0.26  2024-04-12
+### 0.0.26 2024-04-12
 
-* stop testing with sidekiq 5, start testing with sidekiq 7
-* get rid of deprecation warnings from sidekiq 7
-* predictable redis failures (like timeout errors) are now retried (once) instead of being treated like any other error (being logged)
-* instead of immediately starting with processing full batch size, exponential progress is implemented so that big messages OOMing the infrastructure can be pushed through
+- stop testing with sidekiq 5, start testing with sidekiq 7
+- get rid of deprecation warnings from sidekiq 7
+- predictable redis failures (like timeout errors) are now retried (once) instead of being treated like any other error (being logged)
+- instead of immediately starting with processing full batch size, exponential progress is implemented so that big messages OOMing the infrastructure can be pushed through
 
-### 0.0.25  2022-05-27
+### 0.0.25 2022-05-27
 
-* added ORDER BY when cleaning up with limit #1338
+- added ORDER BY when cleaning up with limit #1338
 
 ### 0.0.24
 
-* fixed error with passing `--cleanup-limit` from CLI down to consumer
-* added missing specs for CLI options
-* added simple smoke spec to ensure CLI builds consumer without errors
-* dropped support for rails 5.2
-* dropped support for ruby 2.6
+- fixed error with passing `--cleanup-limit` from CLI down to consumer
+- added missing specs for CLI options
+- added simple smoke spec to ensure CLI builds consumer without errors
+- dropped support for rails 5.2
+- dropped support for ruby 2.6
 
 ### 0.0.23
 

contrib/ruby_event_store-outbox/lib/ruby_event_store/outbox/sidekiq_processor.rb

@@ -20,7 +20,7 @@ def process(record, now)
 
         queue = parsed_record["queue"]
         raise InvalidPayload.new("Missing queue") if queue.nil? || queue.empty?
-        payload = JSON.generate(parsed_record.merge({ "enqueued_at" => now.to_f }))
+        payload = JSON.generate(parsed_record.merge({ "enqueued_at" => record.created_at.to_f }))
 
         redis.call("LPUSH", "queue:#{queue}", payload)
 

contrib/ruby_event_store-outbox/spec/consumer_spec.rb

@@ -36,7 +36,7 @@ module Outbox
           expect(redis.call("LLEN", "queue:default")).to eq(1)
           payload_in_redis = JSON.parse(redis.call("LINDEX", "queue:default", 0))
           expect(payload_in_redis).to include(JSON.parse(record.payload))
-          expect(payload_in_redis["enqueued_at"]).to eq(clock.tick(locking ? 1 : 0).to_f)
+          expect(payload_in_redis["enqueued_at"]).to eq(record.created_at.to_f)
           expect(record.enqueued_at).to eq(clock.tick(locking ? 1 : 0))
           expect(result).to be(true)
           expect(logger_output.string).to include("Sent 1 messages from outbox table")