From 2ca7987aef9b16f67936f674b9de96969d0838fb Mon Sep 17 00:00:00 2001 From: Tom Date: Wed, 15 Oct 2025 15:22:42 +0700 Subject: [PATCH 1/2] feat(redis): add UI/observability metadata and counters - Add `:dead` ZSET for compact failure records (trim by size and optional age). - Increment `:stat:processed` / `:stat:failed` counters. - Server opts: `stats`, `dead_enabled`, `dead_max`, `dead_timeout`, `failure_backtrace_limit`. - Update docs with keys, examples, and configuration. - Add tests covering failure recording, trimming, and counters. --- context/redis-queue.md | 9 +++ guides/redis-queue/readme.md | 33 ++++++++++ .../job/processor/redis/processing_list.rb | 63 +++++++++++++++++++ lib/async/job/processor/redis/server.rb | 24 ++++++- readme.md | 2 +- test/async/job/processor/processing_list.rb | 55 +++++++++++++++- test/async/job/processor/server.rb | 54 ++++++++++++++-- 7 files changed, 230 insertions(+), 10 deletions(-) diff --git a/context/redis-queue.md b/context/redis-queue.md index 5a85b15..947c9ae 100644 --- a/context/redis-queue.md +++ b/context/redis-queue.md @@ -17,3 +17,12 @@ The delayed queue holds jobs that are not meant to be executed immediately but a ## Processing Queue Once a job is dequeued from the ready queue, it enters the processing queue, signifying that it is currently being executed by a worker. The processing queue is crucial for tracking the progress of jobs and for ensuring that jobs can be retried or recovered in case of worker failure. Each worker emits a heartbeat, and if a worker fails to emit a heartbeat within a specified time, any jobs associated with that worker are automatically moved back to the ready queue for reprocessing. + +## Optional UI/Observability Data + +For operational dashboards, the server can (optionally) record failed job entries in a bounded sorted set and increment simple counters. This provides a Sidekiq‑style “morgue” without requiring a SQL database: + +- `:dead` – ZSET of recent failures, newest first, each member a compact JSON; trimmed by size and (optionally) age. +- `:stat:processed` / `:stat:failed` – integer counters. + +These features are enabled by default and configurable via the server constructor. See the Redis Queue guide for configuration and example queries. diff --git a/guides/redis-queue/readme.md b/guides/redis-queue/readme.md index 5a85b15..a5a3577 100644 --- a/guides/redis-queue/readme.md +++ b/guides/redis-queue/readme.md @@ -17,3 +17,36 @@ The delayed queue holds jobs that are not meant to be executed immediately but a ## Processing Queue Once a job is dequeued from the ready queue, it enters the processing queue, signifying that it is currently being executed by a worker. The processing queue is crucial for tracking the progress of jobs and for ensuring that jobs can be retried or recovered in case of worker failure. Each worker emits a heartbeat, and if a worker fails to emit a heartbeat within a specified time, any jobs associated with that worker are automatically moved back to the ready queue for reprocessing. + +## UI/Observability Keys (Optional) + +For dashboards and operational UIs, the server can emit minimal, bounded metadata when jobs fail and when they are processed successfully: + +- `async-job:dead` (ZSET) — failed jobs, newest first. Members are compact JSON with `jid`, `queue`, `class`, `args`, `error_class`, `error_message`, `error_backtrace[]`, `failed_at`. +- `async-job:stat:processed` (STRING) — total number of successfully processed jobs. +- `async-job:stat:failed` (STRING) — total number of failed job executions. + +You can enable and configure this when constructing the server instance: + +```ruby +server = Async::Job::Processor::Redis::Server.new( + delegate, client, + prefix: "async-job", + stats: true, + dead_enabled: true, + dead_max: 1000, + dead_timeout: nil, + failure_backtrace_limit: 10 +) +``` + +Example queries for a UI: + +```bash +ZREVRANGE async-job:dead 0 19 WITHSCORES +GET async-job:stat:processed +GET async-job:stat:failed +SCAN 0 MATCH async-job:processing:* COUNT 100 +``` + +Note: When used via the Active Job adapter, make sure the job executor re-raises exceptions so the processor can observe failures. diff --git a/lib/async/job/processor/redis/processing_list.rb b/lib/async/job/processor/redis/processing_list.rb index 53fafb3..4726b63 100644 --- a/lib/async/job/processor/redis/processing_list.rb +++ b/lib/async/job/processor/redis/processing_list.rb @@ -3,6 +3,8 @@ # Released under the MIT License. # Copyright, 2024-2025, by Samuel Williams. +require "json" + module Async module Job module Processor @@ -112,6 +114,67 @@ def retry(id) Console.warn(self, "Retrying job: #{id}") @client.evalsha(@retry, 2, @pending_key, @ready_list.key, id) end + + # Record a job failure for UI/observability purposes. + # + # Adds a compact JSON blob to the `:dead` sorted set, scored by failure time. + # Optionally enforces a maximum size and time-based trimming. + # Also increments `:stat:failed` if enabled. + # + # @parameter id [String] The job ID that failed. + # @parameter job [Hash] The deserialized job payload. + # @parameter error [Exception] The exception that was raised. + # @parameter dead_max [Integer] Maximum number of failure entries to retain. + # @parameter failure_backtrace_limit [Integer] Maximum number of backtrace lines to retain. + # @parameter dead_timeout [Numeric | nil] If set, remove entries older than this many seconds. + # @parameter stats_enabled [Boolean] Whether to increment the failed counter. + def record_failure(id, job, error, dead_max: 1000, failure_backtrace_limit: 10, dead_timeout: nil, stats_enabled: true) + now = Time.now.to_f + prefix = self.prefix + dead_key = "#{prefix}:dead" + failed_counter = "#{prefix}:stat:failed" + + payload = { + "jid" => id, + "queue" => job["queue_name"], + "class" => job["job_class"], + "args" => job["arguments"], + "error_class" => error.class.name, + "error_message" => error.message.to_s[0, 1024], + "error_backtrace" => Array(error.backtrace).first(failure_backtrace_limit), + "failed_at" => now + } + + json = JSON.dump(payload) + @client.call('ZADD', dead_key, now, json) + + # Optional time-based trim: + if dead_timeout && dead_timeout.to_f > 0 + cutoff = now - dead_timeout.to_f + @client.call('ZREMRANGEBYSCORE', dead_key, '-inf', cutoff) + end + + # Size-based trim: + count = @client.call('ZCARD', dead_key).to_i + if count > dead_max.to_i && dead_max.to_i > 0 + @client.call('ZREMRANGEBYRANK', dead_key, 0, count - dead_max.to_i - 1) + end + + @client.call('INCR', failed_counter) if stats_enabled + end + + # Increment the processed counter for successful job completions. + # @parameter stats_enabled [Boolean] Whether to increment the counter. + def increment_processed(stats_enabled: true) + return unless stats_enabled + @client.call('INCR', "#{self.prefix}:stat:processed") + end + + # Derive the base key prefix, e.g. "async-job" from a processing key like "async-job:processing". + # @returns [String] + def prefix + @key.sub(/:processing\z/, '') + end # Update heartbeat and requeue any abandoned jobs from inactive workers. # @parameter start_time [Float] The start time for calculating uptime. diff --git a/lib/async/job/processor/redis/server.rb b/lib/async/job/processor/redis/server.rb index d3ead22..b662f7a 100644 --- a/lib/async/job/processor/redis/server.rb +++ b/lib/async/job/processor/redis/server.rb @@ -29,7 +29,8 @@ class Server < Generic # @parameter coder [Async::Job::Coder] The job serialization codec. # @parameter resolution [Integer] The resolution in seconds for delayed job processing. # @parameter parent [Async::Task] The parent task for background processing. - def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, parent: nil) + def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, resolution: 10, parent: nil, + stats: true, dead_enabled: true, dead_max: 1000, dead_timeout: nil, failure_backtrace_limit: 10) super(delegate) @id = SecureRandom.uuid @@ -37,6 +38,11 @@ def initialize(delegate, client, prefix: "async-job", coder: Coder::DEFAULT, res @prefix = prefix @coder = coder @resolution = resolution + @stats_enabled = !!stats + @dead_enabled = !!dead_enabled + @dead_max = dead_max + @dead_timeout = dead_timeout + @failure_backtrace_limit = failure_backtrace_limit @job_store = JobStore.new(@client, "#{@prefix}:jobs") @delayed_jobs = DelayedJobs.new(@client, "#{@prefix}:delayed") @@ -129,8 +135,24 @@ def dequeue(parent) job = @coder.load(@job_store.get(id)) @delegate.call(job) @processing_list.complete(id) + @processing_list.increment_processed(stats_enabled: @stats_enabled) rescue => error Console.error(self, "Job failed with error!", id: id, exception: error) + if @dead_enabled + begin + @processing_list.record_failure( + id, + job || {}, + error, + dead_max: @dead_max, + failure_backtrace_limit: @failure_backtrace_limit, + dead_timeout: @dead_timeout, + stats_enabled: @stats_enabled + ) + rescue => e + Console.warn(self, "Failed to record job failure!", id: id, exception: e) + end + end @processing_list.retry(id) end ensure diff --git a/readme.md b/readme.md index 5f1600d..5927873 100644 --- a/readme.md +++ b/readme.md @@ -10,7 +10,7 @@ Please see the [project documentation](https://socketry.github.io/async-job-proc - [Getting Started](https://socketry.github.io/async-job-processor-redis/guides/getting-started/index) - This guide gives you an overview of the `async-job-processor-redis` gem. - - [Redis Queue](https://socketry.github.io/async-job-processor-redis/guides/redis-queue/index) - This guide gives a brief overview of the implementation of the Redis queue. + - [Redis Queue](https://socketry.github.io/async-job-processor-redis/guides/redis-queue/index) - This guide gives a brief overview of the implementation of the Redis queue, including optional UI/observability keys for failure tracking. ## Releases diff --git a/test/async/job/processor/processing_list.rb b/test/async/job/processor/processing_list.rb index ab37ac5..360e462 100644 --- a/test/async/job/processor/processing_list.rb +++ b/test/async/job/processor/processing_list.rb @@ -12,6 +12,7 @@ require "async/job/processor/redis/processing_list" require "async/job/processor/redis/ready_list" require "async/job/processor/redis/job_store" +require "json" describe Async::Job::Processor::Redis::ProcessingList do include Sus::Fixtures::Async::ReactorContext @@ -95,7 +96,7 @@ end end - with "#requeue" do + with "#requeue" do it "can set heartbeat and requeue abandoned jobs" do # Set up abandoned jobs to test requeue functionality dead_server_id = "dead-server-#{SecureRandom.hex(4)}" @@ -135,8 +136,56 @@ # Should return 0 when no jobs are requeued expect(count).to be == 0 - end - end + end + + with "observability helpers" do + it "records failures into :dead and trims to dead_max; increments counters" do + # Unique namespace per test + obs_prefix = "obs-#{SecureRandom.hex(6)}" + list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) + error = begin + raise "boom" + rescue => e + e + end + + # Record first failure + list.record_failure("jid-1", {"queue_name"=>"default","job_class"=>"Demo","arguments"=>[1]}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + + dead_key = "#{obs_prefix}:dead" + stat_failed = "#{obs_prefix}:stat:failed" + + count1 = client.call('ZCARD', dead_key).to_i + expect(count1).to be == 1 + entry = client.call('ZREVRANGE', dead_key, 0, 0)&.first + data = JSON.parse(entry) + expect(data).to have_keys( + 'jid' => be == 'jid-1', + 'error_class' => be == 'RuntimeError', + 'error_message' => be(:include?, 'boom') + ) + + # Failed counter increments + failed_count = (client.call('GET', stat_failed) || '0').to_i + expect(failed_count).to be >= 1 + + # Add two more failures; dead_max=2 so only 2 most recent remain + list.record_failure("jid-2", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + list.record_failure("jid-3", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + + count2 = client.call('ZCARD', dead_key).to_i + expect(count2).to be == 2 + end + + it "increments processed counter" do + obs_prefix = "obs-#{SecureRandom.hex(6)}" + list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) + list.increment_processed(stats_enabled: true) + val = (client.call('GET', "#{obs_prefix}:stat:processed") || '0').to_i + expect(val).to be == 1 + end + end +end with "#start" do it "can requeue an abandoned job" do diff --git a/test/async/job/processor/server.rb b/test/async/job/processor/server.rb index a56075b..fcc16c2 100644 --- a/test/async/job/processor/server.rb +++ b/test/async/job/processor/server.rb @@ -50,8 +50,8 @@ end end - with "a failed job" do - it "can retry a job" do + with "a failed job" do + it "can retry a job" do server.call(job) failed = false @@ -71,9 +71,53 @@ "data" => be == job["data"], ) - expect(failed).to be == true - end - end + expect(failed).to be == true + end + + it "records failure in :dead and increments counters" do + # Enqueue a job which fails once, then succeeds (same as previous test): + server.call(job) + failed = false + + mock(buffer) do |mock| + mock.before(:call) do |job| + unless failed + failed = true + raise "test error for observability" + end + end + end + + # Consume the retried job so the loop progresses: + buffer.pop + + # Give a moment for failure recording to be written: + sleep 0.05 + + client = Async::Redis::Client.new + dead_key = "#{prefix}:dead" + stat_failed = "#{prefix}:stat:failed" + stat_processed = "#{prefix}:stat:processed" + + # Dead set should have at least one entry: + count = client.call('ZCARD', dead_key).to_i + expect(count).to be > 0 + + # Latest entry should include error_class and error_message: + entry = client.call('ZREVRANGE', dead_key, 0, 0)&.first + data = JSON.parse(entry) + expect(data).to have_keys( + 'error_class' => be == 'RuntimeError', + 'error_message' => be(:include?, 'test error for observability') + ) + + # Failed counter should be >= 1, processed >= 1 (after retry succeeds): + failed_count = (client.call('GET', stat_failed) || '0').to_i + processed_count = (client.call('GET', stat_processed) || '0').to_i + expect(failed_count).to be >= 1 + expect(processed_count).to be >= 1 + end + end with "#status_string" do it "returns a string with the current job counts" do From 54403562009125ea6d4cc7588da015e4fb5ccc5b Mon Sep 17 00:00:00 2001 From: Tom Date: Wed, 15 Oct 2025 21:05:57 +0700 Subject: [PATCH 2/2] fix style problem --- guides/redis-queue/readme.md | 14 +-- .../job/processor/redis/processing_list.rb | 20 ++-- test/async/job/processor/processing_list.rb | 102 +++++++++--------- test/async/job/processor/server.rb | 102 +++++++++--------- 4 files changed, 119 insertions(+), 119 deletions(-) diff --git a/guides/redis-queue/readme.md b/guides/redis-queue/readme.md index a5a3577..920b0d2 100644 --- a/guides/redis-queue/readme.md +++ b/guides/redis-queue/readme.md @@ -30,13 +30,13 @@ You can enable and configure this when constructing the server instance: ```ruby server = Async::Job::Processor::Redis::Server.new( - delegate, client, - prefix: "async-job", - stats: true, - dead_enabled: true, - dead_max: 1000, - dead_timeout: nil, - failure_backtrace_limit: 10 + delegate, client, + prefix: "async-job", + stats: true, + dead_enabled: true, + dead_max: 1000, + dead_timeout: nil, + failure_backtrace_limit: 10 ) ``` diff --git a/lib/async/job/processor/redis/processing_list.rb b/lib/async/job/processor/redis/processing_list.rb index 4726b63..3df6ea8 100644 --- a/lib/async/job/processor/redis/processing_list.rb +++ b/lib/async/job/processor/redis/processing_list.rb @@ -114,7 +114,7 @@ def retry(id) Console.warn(self, "Retrying job: #{id}") @client.evalsha(@retry, 2, @pending_key, @ready_list.key, id) end - + # Record a job failure for UI/observability purposes. # # Adds a compact JSON blob to the `:dead` sorted set, scored by failure time. @@ -146,34 +146,34 @@ def record_failure(id, job, error, dead_max: 1000, failure_backtrace_limit: 10, } json = JSON.dump(payload) - @client.call('ZADD', dead_key, now, json) + @client.call("ZADD", dead_key, now, json) # Optional time-based trim: if dead_timeout && dead_timeout.to_f > 0 cutoff = now - dead_timeout.to_f - @client.call('ZREMRANGEBYSCORE', dead_key, '-inf', cutoff) + @client.call("ZREMRANGEBYSCORE", dead_key, "-inf", cutoff) end # Size-based trim: - count = @client.call('ZCARD', dead_key).to_i + count = @client.call("ZCARD", dead_key).to_i if count > dead_max.to_i && dead_max.to_i > 0 - @client.call('ZREMRANGEBYRANK', dead_key, 0, count - dead_max.to_i - 1) + @client.call("ZREMRANGEBYRANK", dead_key, 0, count - dead_max.to_i - 1) end - @client.call('INCR', failed_counter) if stats_enabled + @client.call("INCR", failed_counter) if stats_enabled end - + # Increment the processed counter for successful job completions. # @parameter stats_enabled [Boolean] Whether to increment the counter. def increment_processed(stats_enabled: true) return unless stats_enabled - @client.call('INCR', "#{self.prefix}:stat:processed") + @client.call("INCR", "#{self.prefix}:stat:processed") end - + # Derive the base key prefix, e.g. "async-job" from a processing key like "async-job:processing". # @returns [String] def prefix - @key.sub(/:processing\z/, '') + @key.sub(/:processing\z/, "") end # Update heartbeat and requeue any abandoned jobs from inactive workers. diff --git a/test/async/job/processor/processing_list.rb b/test/async/job/processor/processing_list.rb index 360e462..6899dde 100644 --- a/test/async/job/processor/processing_list.rb +++ b/test/async/job/processor/processing_list.rb @@ -96,7 +96,7 @@ end end - with "#requeue" do + with "#requeue" do it "can set heartbeat and requeue abandoned jobs" do # Set up abandoned jobs to test requeue functionality dead_server_id = "dead-server-#{SecureRandom.hex(4)}" @@ -136,56 +136,56 @@ # Should return 0 when no jobs are requeued expect(count).to be == 0 - end - - with "observability helpers" do - it "records failures into :dead and trims to dead_max; increments counters" do - # Unique namespace per test - obs_prefix = "obs-#{SecureRandom.hex(6)}" - list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) - error = begin - raise "boom" - rescue => e - e - end - - # Record first failure - list.record_failure("jid-1", {"queue_name"=>"default","job_class"=>"Demo","arguments"=>[1]}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) - - dead_key = "#{obs_prefix}:dead" - stat_failed = "#{obs_prefix}:stat:failed" - - count1 = client.call('ZCARD', dead_key).to_i - expect(count1).to be == 1 - entry = client.call('ZREVRANGE', dead_key, 0, 0)&.first - data = JSON.parse(entry) - expect(data).to have_keys( - 'jid' => be == 'jid-1', - 'error_class' => be == 'RuntimeError', - 'error_message' => be(:include?, 'boom') - ) - - # Failed counter increments - failed_count = (client.call('GET', stat_failed) || '0').to_i - expect(failed_count).to be >= 1 - - # Add two more failures; dead_max=2 so only 2 most recent remain - list.record_failure("jid-2", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) - list.record_failure("jid-3", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) - - count2 = client.call('ZCARD', dead_key).to_i - expect(count2).to be == 2 - end - - it "increments processed counter" do - obs_prefix = "obs-#{SecureRandom.hex(6)}" - list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) - list.increment_processed(stats_enabled: true) - val = (client.call('GET', "#{obs_prefix}:stat:processed") || '0').to_i - expect(val).to be == 1 - end - end -end + end + + with "observability helpers" do + it "records failures into :dead and trims to dead_max; increments counters" do + # Unique namespace per test + obs_prefix = "obs-#{SecureRandom.hex(6)}" + list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) + error = begin + raise "boom" + rescue => e + e + end + + # Record first failure + list.record_failure("jid-1", {"queue_name"=>"default","job_class"=>"Demo","arguments"=>[1]}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + + dead_key = "#{obs_prefix}:dead" + stat_failed = "#{obs_prefix}:stat:failed" + + count1 = client.call("ZCARD", dead_key).to_i + expect(count1).to be == 1 + entry = client.call("ZREVRANGE", dead_key, 0, 0)&.first + data = JSON.parse(entry) + expect(data).to have_keys( + "jid" => be == "jid-1", + "error_class" => be == "RuntimeError", + "error_message" => be(:include?, "boom") + ) + + # Failed counter increments + failed_count = (client.call("GET", stat_failed) || "0").to_i + expect(failed_count).to be >= 1 + + # Add two more failures; dead_max=2 so only 2 most recent remain + list.record_failure("jid-2", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + list.record_failure("jid-3", {"queue_name"=>"default"}, error, dead_max: 2, failure_backtrace_limit: 3, stats_enabled: true) + + count2 = client.call("ZCARD", dead_key).to_i + expect(count2).to be == 2 + end + + it "increments processed counter" do + obs_prefix = "obs-#{SecureRandom.hex(6)}" + list = subject.new(client, "#{obs_prefix}:processing", server_id, ready_list, job_store) + list.increment_processed(stats_enabled: true) + val = (client.call("GET", "#{obs_prefix}:stat:processed") || "0").to_i + expect(val).to be == 1 + end + end + end with "#start" do it "can requeue an abandoned job" do diff --git a/test/async/job/processor/server.rb b/test/async/job/processor/server.rb index fcc16c2..4b6657e 100644 --- a/test/async/job/processor/server.rb +++ b/test/async/job/processor/server.rb @@ -50,8 +50,8 @@ end end - with "a failed job" do - it "can retry a job" do + with "a failed job" do + it "can retry a job" do server.call(job) failed = false @@ -68,56 +68,56 @@ # The job was retried: processed_job = buffer.pop expect(processed_job).to have_keys( - "data" => be == job["data"], - ) + "data" => be == job["data"], + ) - expect(failed).to be == true - end - - it "records failure in :dead and increments counters" do - # Enqueue a job which fails once, then succeeds (same as previous test): - server.call(job) - failed = false - - mock(buffer) do |mock| - mock.before(:call) do |job| - unless failed - failed = true - raise "test error for observability" - end - end - end - - # Consume the retried job so the loop progresses: - buffer.pop - - # Give a moment for failure recording to be written: - sleep 0.05 - - client = Async::Redis::Client.new - dead_key = "#{prefix}:dead" - stat_failed = "#{prefix}:stat:failed" - stat_processed = "#{prefix}:stat:processed" - - # Dead set should have at least one entry: - count = client.call('ZCARD', dead_key).to_i - expect(count).to be > 0 - - # Latest entry should include error_class and error_message: - entry = client.call('ZREVRANGE', dead_key, 0, 0)&.first - data = JSON.parse(entry) - expect(data).to have_keys( - 'error_class' => be == 'RuntimeError', - 'error_message' => be(:include?, 'test error for observability') - ) - - # Failed counter should be >= 1, processed >= 1 (after retry succeeds): - failed_count = (client.call('GET', stat_failed) || '0').to_i - processed_count = (client.call('GET', stat_processed) || '0').to_i - expect(failed_count).to be >= 1 - expect(processed_count).to be >= 1 - end - end + expect(failed).to be == true + end + + it "records failure in :dead and increments counters" do + # Enqueue a job which fails once, then succeeds (same as previous test): + server.call(job) + failed = false + + mock(buffer) do |mock| + mock.before(:call) do |job| + unless failed + failed = true + raise "test error for observability" + end + end + end + + # Consume the retried job so the loop progresses: + buffer.pop + + # Give a moment for failure recording to be written: + sleep 0.05 + + client = Async::Redis::Client.new + dead_key = "#{prefix}:dead" + stat_failed = "#{prefix}:stat:failed" + stat_processed = "#{prefix}:stat:processed" + + # Dead set should have at least one entry: + count = client.call("ZCARD", dead_key).to_i + expect(count).to be > 0 + + # Latest entry should include error_class and error_message: + entry = client.call("ZREVRANGE", dead_key, 0, 0)&.first + data = JSON.parse(entry) + expect(data).to have_keys( + "error_class" => be == "RuntimeError", + "error_message" => be(:include?, "test error for observability") + ) + + # Failed counter should be >= 1, processed >= 1 (after retry succeeds): + failed_count = (client.call("GET", stat_failed) || "0").to_i + processed_count = (client.call("GET", stat_processed) || "0").to_i + expect(failed_count).to be >= 1 + expect(processed_count).to be >= 1 + end + end with "#status_string" do it "returns a string with the current job counts" do