Add Redis observability metadata for a simple UI (dead set + counters)

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:
+
+- `<prefix>:dead` – ZSET of recent failures, newest first, each member a compact JSON; trimmed by size and (optionally) age.
+- `<prefix>:stat:processed` / `<prefix>: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.

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.

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
@@ -113,6 +115,67 @@ def retry(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 `<prefix>:dead` sorted set, scored by failure time.
+					# Optionally enforces a maximum size and time-based trimming.
+					# Also increments `<prefix>: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.
 					# @parameter delay [Numeric] The heartbeat update interval.

lib/async/job/processor/redis/server.rb

@@ -29,14 +29,20 @@ 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
 						@client = client
 						@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

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
 

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
@@ -136,6 +137,54 @@
 			# Should return 0 when no jobs are requeued
 			expect(count).to be == 0
 		end
+
+		with "observability helpers" do
+			it "records failures into <prefix>: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

test/async/job/processor/server.rb

@@ -68,11 +68,55 @@
 			# 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 <prefix>: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