From 5132eb0cb0a9626262e0f575fbb5ede3858f8761 Mon Sep 17 00:00:00 2001 From: Alex Rohachev Date: Mon, 5 Oct 2026 11:17:23 +0000 Subject: [PATCH 1/4] [Deferred] Add dead task listing and inspection Co-authored-by: Serhii Sadovskyi --- lib/rage/deferred/backends/disk.rb | 275 ++++--- lib/rage/deferred/backends/nil.rb | 11 +- lib/rage/deferred/dead_task.rb | 107 +++ lib/rage/deferred/dead_tasks.rb | 49 ++ lib/rage/deferred/deferred.rb | 23 +- spec/deferred/backends/disk_spec.rb | 728 +++++++++++++++++- spec/deferred/backends/nil_spec.rb | 13 + spec/deferred/dead_tasks_spec.rb | 389 ++++++++++ spec/spec_helper.rb | 1 + .../shared_examples/dead_task_exact_lookup.rb | 50 ++ 10 files changed, 1548 insertions(+), 98 deletions(-) create mode 100644 lib/rage/deferred/dead_task.rb create mode 100644 lib/rage/deferred/dead_tasks.rb create mode 100644 spec/deferred/backends/nil_spec.rb create mode 100644 spec/deferred/dead_tasks_spec.rb create mode 100644 spec/support/shared_examples/dead_task_exact_lookup.rb diff --git a/lib/rage/deferred/backends/disk.rb b/lib/rage/deferred/backends/disk.rb index 8faba70a..9c328b9d 100644 --- a/lib/rage/deferred/backends/disk.rb +++ b/lib/rage/deferred/backends/disk.rb @@ -10,8 +10,9 @@ # * `remove_task` - called when a task has to be removed from the storage; # * `pending_tasks` - the method should iterate over the underlying storage and return a list of tasks to replay; # * `add_dead_task` - called when a task has exhausted its retries or aborted them; -# * `list_dead_tasks` - return a list of dead tasks, newest first; -# * `find_dead_task` - return a single dead task; +# * `each_dead_task` - calls the block once for each logical dead task in a fixed snapshot, +# ordered by the physical positions of the selected winning records; +# * `find_dead_task` - returns the newest frame-valid dead-task record with the exact outer ID; # * `remove_dead_tasks` - permanently delete dead tasks; # class Rage::Deferred::Backends::Disk @@ -53,19 +54,22 @@ def add_dead_task(task_id, context, exception, task_class:, attempts:) @dead_tasks_storage.add(task_id, context, exception, task_class:, attempts:) end - # Return a list of dead-lettered tasks, newest first. - # @param limit [Integer, nil] the maximum number of records to return - # @param offset [Integer] the number of records to skip - # @return [Array] - def list_dead_tasks(limit: nil, offset: 0) - @dead_tasks_storage.list(limit:, offset:) + # Call the block once for each logical dead task in a fixed snapshot. + # Selected winners are ordered by their physical positions from oldest to newest. + # Changes to the live storage during iteration do not change the selected records. + # @yieldparam record [Hash] a decoded selected dead-task record + # @return [Rage::Deferred::Backends::Disk::DeadTasksStorage] + # @private + def each_dead_task(&block) + @dead_tasks_storage.each(&block) end - # Return a single dead-lettered task. - # @param id [String] the id of the dead-task record - # @return [Hash, nil] + # Return the newest frame-valid dead-task record with the exact outer ID. + # @param id [String] the persisted task ID + # @return [Hash, nil] the decoded record, or nil if there is no matching record + # @private def find_dead_task(id) - @dead_tasks_storage.find(id) + @dead_tasks_storage.find_by_id(id) end # Permanently delete dead-lettered tasks. @@ -421,25 +425,55 @@ def add(task_id, context, exception, task_class:, attempts:) end end - # Return dead task records, newest first. - # @param limit [Integer, nil] maximum number of records to return, or all records if nil - # @param offset [Integer] number of newest records to skip - # @return [Array] task records that could be read successfully + # Call the block once for each selected dead-task record, oldest winning position first. + # + # The method opens the current file while it holds the lock. + # It sets the read limit after the last complete record and then releases the lock. + # Records added after this point are not part of the traversal. + # After releasing the lock, it scans complete records sequentially forwards and selects + # the newest frame-valid record for each authoritative outer task ID. Repeated IDs are + # ordered at their winning records' physical positions. It then seeks to, decodes, and + # yields one selected payload at a time, without decoding the next winner before control + # returns from the current yield. + # Each call has separate traversal state. + # + # @yieldparam record [Hash] a decoded dead-task record + # @return [self] # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked - def list(limit: nil, offset: 0) - records = read_records.reverse - records = records.drop(offset) if offset > 0 - records = records.first(limit) if limit + # @raise [SystemCallError] when an operation on the selected storage file fails + def each + with_snapshot do |storage, snapshot_end| + records_index(storage, snapshot_end).each do |task_id, (offset, length)| + storage.seek(offset, IO::SEEK_SET) + yield Marshal.load(storage.read(length).undump).merge(id: task_id) + end - records + self + end end - # Find a dead task by its id. - # @param id [String] persisted task id - # @return [Hash, nil] the task record, or nil when no record matches + # Return the newest frame-valid record with the exact authoritative outer ID. + # Scan the fixed complete-record snapshot forwards without decoding candidate payloads. + # Decode only the newest matching payload after scanning reaches the snapshot boundary. + # @param id [String] the persisted task ID + # @return [Hash, nil] the decoded record, or nil if there is no matching record # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked - def find(id) - read_records.find { |record| record[:id] == id } + # @raise [SystemCallError] when an operation on the selected storage file fails + def find_by_id(id) + with_snapshot do |storage, snapshot_end| + serialized_record_location = nil + each_framed_record(storage, snapshot_end) do |task_id, offset, length| + next unless task_id == id + + serialized_record_location = [offset, length] + end + + next unless serialized_record_location + + offset, length = serialized_record_location + storage.seek(offset, IO::SEEK_SET) + Marshal.load(storage.read(length).undump).merge(id: id) + end end # Permanently delete the records with the given ids. @@ -459,7 +493,7 @@ def remove(ids) File.open(@tmp_storage_path, File::WRONLY | File::CREAT | File::TRUNC | File::BINARY, 0o644) do |tmp| File.open(@storage_path, File::RDONLY | File::BINARY) do |storage| storage.each_line do |entry| - id = entry_id(entry) + id, = framed_record(entry.chomp) if id.nil? next # drop corrupted records; they can neither be listed nor deleted otherwise @@ -489,95 +523,145 @@ def remove(ids) private - # Persist changes to the live file's directory entry, including file creation and replacement. - # @return [void] - def sync_storage_directory - File.open(@storage_path.dirname, File::RDONLY, &:fsync) + # Index the newest frame-valid serialized-payload location for each authoritative outer ID. + # Deleting and reinserting a repeated ID moves it to its winning record's physical position, + # so Hash insertion order visits the completed index from oldest winner to newest winner. + # @param storage [File] the descriptor for the fixed snapshot + # @param snapshot_end [Integer] the byte boundary after its last complete record + # @return [Hash{String => Array<(Integer, Integer)>}] ordered serialized-payload locations + def records_index(storage, snapshot_end) + records_index = {} + each_framed_record(storage, snapshot_end) do |task_id, offset, length| + records_index.delete(task_id) + records_index[task_id] = [offset, length].freeze + end + + records_index end - # Remove an incomplete final entry left by an interrupted append. - # @param storage [File] store opened for reading and writing + # Scan complete physical records forwards through a fixed snapshot boundary. + # Silently skip malformed frames and invalid CRCs. Do not decode or inspect serialized payloads. + # @param storage [File] the descriptor for the fixed snapshot + # @param snapshot_end [Integer] the byte boundary after its last complete record + # @yieldparam task_id [String] the authoritative outer task ID + # @yieldparam offset [Integer] the serialized payload's byte offset + # @yieldparam length [Integer] the serialized payload's byte length # @return [void] - def repair_torn_tail(storage) + # @private + def each_framed_record(storage, snapshot_end) + position = 0 + storage.seek(0, IO::SEEK_SET) + + while position < snapshot_end + offset = position + entry = storage.gets + break unless entry + + position += entry.bytesize + length = entry.bytesize - 1 + entry = entry.byteslice(0, length) + task_id, serialized_record = framed_record(entry) + next unless task_id + + serialized_record_offset = offset + length - serialized_record.bytesize + yield task_id, serialized_record_offset, serialized_record.bytesize + end + end + + # Open the live file and capture its last complete-record boundary under the permanent lock. + # Release the lock before scanning, decoding, or yielding. The retained descriptor continues + # to identify the same snapshot file if the live path is replaced while the block runs. + # Close the descriptor on unwind; a close error does not replace an active exception. + # @yieldparam storage [File] the descriptor for the fixed snapshot + # @yieldparam snapshot_end [Integer] the fixed end position of the last complete record + # @yieldreturn [Object] the result from the read operation + # @return [Object] the block result + def with_snapshot + storage = nil + active_exception = nil + + begin + snapshot_end = with_lock("read tasks from") do + storage = File.open(@storage_path, File::RDONLY | File::BINARY) + complete_record_end(storage) + end + + yield storage, snapshot_end + rescue Exception => e + active_exception = e + raise + ensure + begin + storage.close if storage && !storage.closed? + rescue + raise unless active_exception + end + end + end + + # Locate the byte boundary immediately after the last newline-terminated record. + # If the file has an incomplete tail, scan backwards in bounded chunks without parsing records. + # Do not modify the file; return zero when it contains no complete record. + # @param storage [File] an open dead-task file + # @return [Integer] the byte boundary after the last complete record + def complete_record_end(storage) storage.seek(0, IO::SEEK_END) end_position = storage.pos - return if end_position == 0 + return 0 if end_position == 0 storage.seek(-1, IO::SEEK_END) - return if storage.read(1) == "\n" + return end_position if storage.read(1) == "\n" position = end_position - truncate_at = 0 - while position > 0 chunk_start = [position - TAIL_SCAN_CHUNK_SIZE, 0].max storage.seek(chunk_start, IO::SEEK_SET) chunk = storage.read(position - chunk_start) if (newline_index = chunk.rindex("\n")) - truncate_at = chunk_start + newline_index + 1 - break + return chunk_start + newline_index + 1 end position = chunk_start end - storage.truncate(truncate_at) + 0 end - # Read valid records, keeping only the latest entry for each task id. - # Corrupted or unreadable records are reported and skipped. - # @return [Array] - # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked - def read_records - entries = with_lock("read tasks from") do - result, corrupted_count = {}, 0 - - File.open(@storage_path, File::RDONLY | File::BINARY) do |storage| - storage.each_line(chomp: true) do |entry| - id = entry_id(entry) - - if id.nil? - corrupted_count += 1 - next - end - - # the same task can be dead-lettered more than once if the worker crashed - # before the task was removed from the write-ahead log - result.delete(id) - result[id] = entry - end - end + # Check only the outer storage frame and return its authoritative ID and serialized payload. + # Return nil for a malformed frame, invalid CRC, unexpected operation, or empty outer ID. + # Do not decode the payload or validate its shape or any ID stored inside it. + # @param entry [String] one complete physical record without the newline + # @return [Array<(String, String)>, nil] + def framed_record(entry) + return unless entry.bytesize > ENTRY_CRC_HEX_WIDTH + 1 + return unless entry.getbyte(ENTRY_CRC_HEX_WIDTH) == 58 # `:` + + signature = entry.byteslice(0, ENTRY_CRC_HEX_WIDTH) + payload = entry.byteslice((ENTRY_CRC_HEX_WIDTH + 1)..) + operation_prefix = "#{ENTRY_OP}:" + return unless payload.start_with?(operation_prefix) + return unless signature == Zlib.crc32(payload).to_s(16).rjust(ENTRY_CRC_HEX_WIDTH, "0") - if corrupted_count != 0 - puts "WARNING: Detected #{corrupted_count} corrupted dead-lettered task(s)" - end + id_start = operation_prefix.bytesize + separator_index = payload.index(":", id_start) + return unless separator_index && separator_index > id_start - result - end - - entries.filter_map do |id, entry| - _, _, _, serialized_record = entry.split(":", 4) - Marshal.load(serialized_record.undump) - rescue ArgumentError, NameError, TypeError => e - puts "ERROR: Can't deserialize the dead-lettered task with id #{id}: (#{e.class}) #{e.message}" - nil - end + [payload.byteslice(id_start...separator_index), payload.byteslice((separator_index + 1)..)] end - # Validate a stored entry and extract its task id. - # @param entry [String] serialized entry, optionally ending with a newline - # @return [String, nil] task id, or nil if the entry is malformed or corrupted - def entry_id(entry) - entry = entry.chomp - - signature, payload = entry[0...ENTRY_CRC_HEX_WIDTH], entry[(ENTRY_CRC_HEX_WIDTH + 1)..] - return if signature.nil? || payload.nil? || !payload.start_with?("#{ENTRY_OP}:") - return unless signature == Zlib.crc32(payload).to_s(16).rjust(ENTRY_CRC_HEX_WIDTH, "0") + # Call `fsync` on the storage directory after file creation or replacement. + # @return [void] + def sync_storage_directory + File.open(@storage_path.dirname, File::RDONLY, &:fsync) + end - id_start = ENTRY_CRC_HEX_WIDTH + 1 + ENTRY_OP.length + 1 - separator_index = entry.index(":", id_start) - entry[id_start...separator_index] if separator_index + # Remove bytes after the last complete record. An interrupted append can leave these bytes. + # @param storage [File] the store opened for reading and writing + # @return [void] + def repair_torn_tail(storage) + truncate_at = complete_record_end(storage) + storage.truncate(truncate_at) if truncate_at < storage.size end # Serialize a task record as a checksummed, newline-delimited entry. @@ -611,12 +695,21 @@ def with_lock(operation) end @locked = true + active_exception = nil begin yield + rescue Exception => e + active_exception = e + raise ensure - @lock_file.flock(File::LOCK_UN) - @locked = false + begin + @lock_file.flock(File::LOCK_UN) + rescue + raise unless active_exception + ensure + @locked = false + end end end end diff --git a/lib/rage/deferred/backends/nil.rb b/lib/rage/deferred/backends/nil.rb index 25829019..ab75a932 100644 --- a/lib/rage/deferred/backends/nil.rb +++ b/lib/rage/deferred/backends/nil.rb @@ -17,10 +17,17 @@ def pending_tasks def add_dead_task(_, _, _, **) end - def list_dead_tasks(**) - [] + # Yield no dead-task records. + # @return [self] + # @private + def each_dead_task + self end + # Return no dead task for an exact ID lookup. + # @param _ [String] the persisted task ID + # @return [nil] + # @private def find_dead_task(_) end diff --git a/lib/rage/deferred/dead_task.rb b/lib/rage/deferred/dead_task.rb new file mode 100644 index 00000000..8aa43d7c --- /dev/null +++ b/lib/rage/deferred/dead_task.rb @@ -0,0 +1,107 @@ +# frozen_string_literal: true + +## +# A read-only view of a deferred task that used all its retries or stopped retrying. +# +# Values come from trusted local dead-task storage. +# Reading metadata does not load the task class or deserialize the stored execution context. +# {#args} and {#kwargs} load the context only when you call them. +# The context is trusted local Marshal data. +# Argument data can contain credentials or personal data. Protect this data from unauthorized access. +class Rage::Deferred::DeadTask + # @return [String] the task's persisted identifier + attr_reader :id + + # @return [String] the stored task class name + attr_reader :task_class + + # @return [Integer] the number of attempts to process the task + attr_reader :attempts + + # @return [Time] the time encoded in the task's persisted identifier + attr_reader :enqueued_at + + # @return [Time] the time when the final attempt failed + attr_reader :failed_at + + # @return [String] the stored class name of the final exception + attr_reader :exception_class + + # @return [String] the stored message of the final exception + attr_reader :exception_message + + # @return [Array] the frozen stored backtrace of the final exception + attr_reader :backtrace + + # @private + def initialize(record) + @id = record[:id].dup.freeze + @task_class = record[:task_class].dup.freeze + @attempts = record[:attempts] + @enqueued_at = Time.at(record[:enqueued_at]) + @failed_at = Time.at(record[:failed_at]) + + @exception_class = record[:exception_class].dup.freeze + @exception_message = record[:exception_message].dup.freeze + @backtrace = record[:backtrace].map { |line| line.dup.freeze }.freeze + @__context = record[:context].dup.freeze + end + + # Return the original positional arguments. + # + # The first successful call decodes the trusted local Marshal context. + # The method freezes the decoded object graph and returns the same Array on later calls. + # Changes to the returned Array cannot change the stored replay input. + # A stored `nil` produces an empty Array. + # The method does not cache a decoding failure. + # You can try again after you restore the application constants. + # + # Argument data can contain credentials or personal data. Protect this data from unauthorized access. + # + # @return [Array] frozen positional arguments + # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded + def args + decode_context unless defined?(@__args) + @__args + end + + # Return the original keyword arguments. + # + # The first successful call decodes the trusted local Marshal context. + # The method freezes the decoded object graph and returns the same Hash on later calls. + # Changes to the returned Hash cannot change the stored replay input. + # A stored `nil` produces an empty Hash. + # This behavior keeps positional and keyword arguments separate on Ruby 3.3. + # The method does not cache a decoding failure. + # You can try again after you restore the application constants. + # + # Argument data can contain credentials or personal data. Protect this data from unauthorized access. + # + # @return [Hash] frozen keyword arguments + # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded + def kwargs + decode_context unless defined?(@__kwargs) + @__kwargs + end + + private + + # @private + def decode_context + context = Marshal.load(@__context, freeze: true) + args = Rage::Deferred::Context.get_args(context) + kwargs = Rage::Deferred::Context.get_kwargs(context) + args = [].freeze if args.nil? + kwargs = {}.freeze if kwargs.nil? + + @__args = args + @__kwargs = kwargs + rescue => e + error = Rage::Deferred::DeadTaskContextDeserializationError.new( + "Could not deserialize execution context for dead task #{@id}" + ) + raise error, cause: e + end + + private_class_method :new +end diff --git a/lib/rage/deferred/dead_tasks.rb b/lib/rage/deferred/dead_tasks.rb new file mode 100644 index 00000000..f703c156 --- /dev/null +++ b/lib/rage/deferred/dead_tasks.rb @@ -0,0 +1,49 @@ +# frozen_string_literal: true + +## +# The collection of deferred tasks that used all their retries or stopped retrying. +# +# The first call to {Rage::Deferred.dead_tasks} creates a collection wrapper around the configured backend. +# Later calls return the same wrapper. +# The wrapper uses the same backend object as the deferred queue. +# The wrapper does not keep traversal state. +# Each traversal requests records from the configured backend. +class Rage::Deferred::DeadTasks + include Enumerable + + # @private + def initialize(backend) + @backend = backend + end + + # Call the block for each read-only dead-task summary. Start with the oldest task. + # + # Without a block, this method returns a standard lazy Ruby Enumerator. + # + # @yieldparam dead_task [Rage::Deferred::DeadTask] a read-only task summary + # @return [Enumerator, self] a new Enumerator without a block, otherwise this collection + def each + return enum_for(__method__) unless block_given? + + @backend.each_dead_task do |record| + yield Rage::Deferred::DeadTask.send(:new, record) + end + + self + end + + # Return the dead task with the exact persisted ID. + # + # The ID must be a String. The method does not convert other values to a String. + # The method does not load the stored task class or deserialize its execution context. + # + # @param id [String] the exact persisted task ID + # @return [Rage::Deferred::DeadTask, nil] the matching task, or nil + # @raise [TypeError] when the ID is not a String + def find_by_id(id) + raise TypeError, "dead task id must be a String" unless id.is_a?(String) + + record = @backend.find_dead_task(id) + Rage::Deferred::DeadTask.send(:new, record) if record + end +end diff --git a/lib/rage/deferred/deferred.rb b/lib/rage/deferred/deferred.rb index 00533fac..5fe2a06f 100644 --- a/lib/rage/deferred/deferred.rb +++ b/lib/rage/deferred/deferred.rb @@ -51,6 +51,18 @@ def self.wrap(instance, delay: nil, delay_until: nil) Rage::Deferred::Proxy.new(instance, delay:, delay_until:) end + # Return the collection of tasks that used all their retries or stopped retrying. + # + # The method returns the same collection for each call. + # The collection uses the same backend object as the deferred queue. + # The first call can initialize the backend. + # Obtaining the collection does not open, scan, or decode dead-task records. + # + # @return [Rage::Deferred::DeadTasks] + def self.dead_tasks + @__dead_tasks ||= Rage::Deferred::DeadTasks.new(__backend) + end + # @private def self.__backend @__backend ||= Rage.config.deferred.backend @@ -100,11 +112,20 @@ module Backends class PushTimeout < StandardError end - # Raised when a dead-tasks store operation cannot acquire the lock within the retry budget. + # Raised when a dead-task store operation cannot get the lock after all retries. class DeadTasksLockTimeout < StandardError end + + # Raised when the method cannot decode the stored execution context of a dead task. + # + # The error message identifies the dead task. + # `Exception#cause` contains the original Marshal error or context-layout error. + class DeadTaskContextDeserializationError < StandardError + end end +require_relative "dead_task" +require_relative "dead_tasks" require_relative "task" require_relative "scheduler" require_relative "queue" diff --git a/spec/deferred/backends/disk_spec.rb b/spec/deferred/backends/disk_spec.rb index a47023fa..f41387e6 100644 --- a/spec/deferred/backends/disk_spec.rb +++ b/spec/deferred/backends/disk_spec.rb @@ -325,7 +325,7 @@ def stored_records add_dead_task("2-2-2") - expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["2-2-2", "1-1-1"]) + expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1", "2-2-2"]) end it "repairs a file containing only an incomplete entry before appending" do @@ -334,7 +334,7 @@ def stored_records add_dead_task("1-1-1") - expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["1-1-1"]) + expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1"]) end it "repairs a torn tail longer than one scan chunk" do @@ -409,13 +409,733 @@ def stored_records end end + describe "#each_dead_task" do + def each_dead_task + backend.enum_for(:each_dead_task).to_a + end + + def finish_external_traversal(iterator) + records = [] + while true + records << iterator.next + end + rescue StopIteration + records + end + + def append_physical_record(outer_id, record) + entry = dead_tasks_storage.send(:build_entry, outer_id, record) + dead_tasks_path.open("ab") { |storage| storage.write(entry) } + end + + def append_serialized_record(outer_id, serialized_record) + payload = "dead_task:#{outer_id}:#{serialized_record}" + crc = Zlib.crc32(payload).to_s(16).rjust(8, "0") + dead_tasks_path.open("ab") { |storage| storage.write("#{crc}:#{payload}\n") } + end + + it "keeps backend traversal argument-free" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + add_dead_task("1700000003-1-3") + + expect(backend.method(:each_dead_task).parameters).to eq([[:block, :block]]) + expect { backend.each_dead_task(limit: 1) {} }.to raise_error(ArgumentError) + expect(each_dead_task.map { |record| record[:id] }).to eq( + ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"] + ) + end + + it "yields the newest frame-valid duplicate" do + add_dead_task("1700000001-1-1") + older = stored_records.last + append_physical_record("1700000001-1-1", older.merge(task_class: "NewTask", failed_at: older[:failed_at] + 1)) + + records = each_dead_task + + expect(records.length).to eq(1) + expect(records.first[:task_class]).to eq("NewTask") + end + + it "orders a duplicate logical task at its winning record's position" do + add_dead_task("1700000001-1-1") + original = stored_records.last + add_dead_task("1700000002-1-2") + append_physical_record("1700000001-1-1", original.merge(task_class: "MovedTask")) + + records = each_dead_task + + expect(records.map { |record| record[:id] }).to eq(["1700000002-1-2", "1700000001-1-1"]) + expect(records.last[:task_class]).to eq("MovedTask") + end + + it "falls back to an older duplicate only when newer frames are physically invalid" do + add_dead_task("1700000001-1-1") + dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:1700000001-1-1:garbage\n") } + + records = each_dead_task + + expect(records.length).to eq(1) + expect(records.first[:id]).to eq("1700000001-1-1") + expect(records.first[:task_class]).to eq("SendWelcomeEmail") + end + + it "does not fall back when the newest frame-valid payload cannot be decoded" do + add_dead_task("1700000001-1-1") + append_serialized_record("1700000001-1-1", "not-a-dumped-string") + + expect { each_dead_task }.to raise_error(RuntimeError, /dumped string/) + end + + it "silently skips malformed frames and an incomplete tail" do + add_dead_task("1700000001-1-1") + dead_tasks_path.open("ab") do |storage| + storage.write("bad\n") + storage.write("00000000:not_dead_task:id:payload\n") + storage.write("deadbeef:dead_task:1700000002-1-2:secret-payload\n") + storage.write("incomplete-secret-tail") + end + snapshot_bytes = dead_tasks_path.binread + + expect { + expect(each_dead_task.map { |record| record[:id] }).to eq(["1700000001-1-1"]) + }.not_to output.to_stdout + expect { each_dead_task }.not_to output.to_stderr + expect(dead_tasks_path.binread).to eq(snapshot_bytes) + end + + it "does not schema-check frame-valid payloads during selection" do + add_dead_task("1700000001-1-1") + valid = stored_records.last + append_physical_record("1700000001-1-1", valid.merge(id: "inner-id", attempts: "three")) + + expect(each_dead_task).to contain_exactly(include(id: "1700000001-1-1", attempts: "three")) + end + + it "does not deserialize opaque context bytes" do + add_dead_task("1700000001-1-1") + record = stored_records.last.merge(context: "not a Marshal payload") + append_physical_record("1700000002-1-2", record.merge(id: "1700000002-1-2")) + + expect(each_dead_task.map { |entry| entry[:id] }).to eq(["1700000001-1-1", "1700000002-1-2"]) + end + + it "reads large physical records during the forward scan" do + add_dead_task("1700000001-1-1") + record = stored_records.last + append_physical_record( + "1700000002-1-2", + record.merge(id: "1700000002-1-2", exception_message: "x" * 20_000) + ) + + expect(each_dead_task.map { |entry| entry[:id] }).to eq(["1700000001-1-1", "1700000002-1-2"]) + end + + it "returns and decodes only the oldest 20 winning records for first and take" do + 25.times { |index| add_dead_task("170000#{index.to_s.rjust(4, "0")}-1-1") } + decode_count = 0 + allow(Marshal).to receive(:load).and_wrap_original do |original, *args| + decode_count += 1 + original.call(*args) + end + + expected_ids = 20.times.map { |index| "170000#{index.to_s.rjust(4, "0")}-1-1" } + + expect(backend.enum_for(:each_dead_task).first(20).map { |record| record[:id] }).to eq(expected_ids) + expect(decode_count).to eq(20) + + decode_count = 0 + expect(backend.enum_for(:each_dead_task).take(20).map { |record| record[:id] }).to eq(expected_ids) + expect(decode_count).to eq(20) + end + + it "finishes frame-only winner selection before the first yield and decodes only the consumed winner" do + 5.times { |index| add_dead_task("170000000#{index}-1-1") } + scanned_ids = [] + allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |original, entry| + original.call(entry).tap { |framed_record| scanned_ids << framed_record&.first } + end + allow(Marshal).to receive(:load).and_wrap_original do |original, *args| + expect(scanned_ids).to eq(5.times.map { |index| "170000000#{index}-1-1" }) + original.call(*args) + end + + first = backend.enum_for(:each_dead_task).first + + expect(first[:id]).to eq("1700000000-1-1") + expect(Marshal).to have_received(:load).once + end + + it "decodes each winner immediately before yielding it" do + 3.times { |index| add_dead_task("170000000#{index}-1-1") } + decoded_ids = [] + yielded_ids = [] + allow(Marshal).to receive(:load).and_wrap_original do |original, *args| + original.call(*args).tap { |record| decoded_ids << record[:id] } + end + + backend.each_dead_task do |record| + yielded_ids << record[:id] + expect(decoded_ids).to eq(yielded_ids) + end + end + + it "does not open or inspect a snapshot until first advancement" do + add_dead_task("1700000001-1-1") + allow(dead_tasks_storage).to receive(:with_snapshot).and_call_original + allow(Marshal).to receive(:load).and_call_original + + iterator = backend.enum_for(:each_dead_task) + + expect(dead_tasks_storage).not_to have_received(:with_snapshot) + expect(Marshal).not_to have_received(:load) + + iterator.next + + expect(dead_tasks_storage).to have_received(:with_snapshot).once + expect(Marshal).to have_received(:load).once + end + + it "excludes an incomplete tail and records appended after first advancement" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:partial") } + iterator = backend.enum_for(:each_dead_task) + + expect(iterator.next[:id]).to eq("1700000001-1-1") + add_dead_task("1700000003-1-3") + + expect(finish_external_traversal(iterator).map { |record| record[:id] }).to eq(["1700000002-1-2"]) + expect(each_dead_task.map { |record| record[:id] }).to eq( + ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"] + ) + end + + it "keeps the captured inode stable across rename-based compaction" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + add_dead_task("1700000003-1-3") + iterator = backend.enum_for(:each_dead_task) + + expect(iterator.next[:id]).to eq("1700000001-1-1") + backend.remove_dead_tasks("1700000002-1-2") + + expect(finish_external_traversal(iterator).map { |record| record[:id] }).to eq( + ["1700000002-1-2", "1700000003-1-3"] + ) + expect(each_dead_task.map { |record| record[:id] }).to eq(["1700000001-1-1", "1700000003-1-3"]) + end + + it "releases the permanent lock before decoding and yielding" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + allow(Marshal).to receive(:load).and_wrap_original do |original, *args| + expect(dead_tasks_storage.instance_variable_get(:@locked)).to eq(false) + original.call(*args) + end + + backend.each_dead_task do + expect(dead_tasks_storage.instance_variable_get(:@locked)).to eq(false) + File.open(lock_path, File::WRONLY) do |lock| + expect(lock.flock(File::LOCK_EX | File::LOCK_NB)).to be_truthy + lock.flock(File::LOCK_UN) + end + end + end + + it "closes the snapshot after exhaustion, block exit, and terminal Enumerable exits" do + add_dead_task("1700000001-1-1") + descriptors = [] + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptors << storage + operation.call(storage, snapshot_end) + end + end + collection = Rage::Deferred::DeadTasks.new(backend) + + backend.each_dead_task {} + backend.each_dead_task { break } + catch(:finished) { backend.each_dead_task { throw :finished } } + collection.find { true } + collection.first + collection.take(1) + + expect(descriptors.length).to eq(6) + expect(descriptors).to all(be_closed) + end + + it "keeps a partial external traversal open until it is exhausted" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + operation.call(storage, snapshot_end) + end + end + iterator = backend.enum_for(:each_dead_task) + + iterator.next + expect(descriptor).not_to be_closed + + finish_external_traversal(iterator) + expect(descriptor).to be_closed + end + + it "does not let cleanup replace an active user exception" do + add_dead_task("1700000001-1-1") + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:close).and_wrap_original do |close| + close.call + raise Errno::EIO + end + operation.call(storage, snapshot_end) + end + end + + expect { + backend.each_dead_task { raise "user failure" } + }.to raise_error(RuntimeError, "user failure") + expect(descriptor).to be_closed + end + + it "propagates a descriptor cleanup error when it is the only failure" do + add_dead_task("1700000001-1-1") + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + allow(storage).to receive(:close).and_wrap_original do |close| + close.call + raise Errno::EIO + end + operation.call(storage, snapshot_end) + end + end + + expect { + backend.each_dead_task {} + }.to raise_error(Errno::EIO) + end + + it "propagates a descriptor cleanup error inside an unrelated rescue" do + add_dead_task("1700000001-1-1") + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + allow(storage).to receive(:close).and_wrap_original do |close| + close.call + raise Errno::EIO + end + operation.call(storage, snapshot_end) + end + end + + begin + raise "unrelated failure" + rescue RuntimeError + expect { + backend.each_dead_task {} + }.to raise_error(Errno::EIO) + end + end + + it "propagates a lock cleanup error inside an unrelated rescue" do + lock_file = dead_tasks_storage.instance_variable_get(:@lock_file) + allow(lock_file).to receive(:flock).and_wrap_original do |flock, operation| + flock.call(operation).tap do + raise Errno::EIO if operation == File::LOCK_UN + end + end + + begin + raise "unrelated failure" + rescue RuntimeError + expect { + backend.each_dead_task {} + }.to raise_error(Errno::EIO) + end + end + + it "propagates lazy snapshot-acquisition failures and closes the opened descriptor" do + descriptor = nil + allow(dead_tasks_storage).to receive(:complete_record_end) do |storage| + descriptor = storage + raise Errno::EACCES + end + iterator = backend.enum_for(:each_dead_task) + + expect(descriptor).to be_nil + expect { iterator.next }.to raise_error(Errno::EACCES) + expect(descriptor).to be_closed + end + + it "propagates an open failure from first advancement" do + backend + allow(File).to receive(:open).and_call_original + allow(File).to receive(:open).with(dead_tasks_path, File::RDONLY | File::BINARY).and_raise(Errno::EACCES) + iterator = backend.enum_for(:each_dead_task) + + expect { iterator.next }.to raise_error(Errno::EACCES) + end + + it "propagates forward-scan seek failures and closes the snapshot" do + add_dead_task("1700000001-1-1") + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:seek).and_raise(Errno::EIO) + operation.call(storage, snapshot_end) + end + end + + expect { + backend.each_dead_task {} + }.to raise_error(Errno::EIO) + expect(descriptor).to be_closed + end + + it "propagates forward-scan read failures and closes the snapshot" do + add_dead_task("1700000001-1-1") + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:gets).and_raise(Errno::EIO) + operation.call(storage, snapshot_end) + end + end + + expect { + backend.each_dead_task {} + }.to raise_error(Errno::EIO) + expect(descriptor).to be_closed + end + + it "propagates snapshot lock timeouts on first advancement" do + stub_const("#{described_class}::DeadTasksStorage::LOCK_MAX_ATTEMPTS", 1) + dead_tasks_storage.instance_variable_set(:@locked, true) + iterator = backend.enum_for(:each_dead_task) + + expect { iterator.next }.to raise_error(Rage::Deferred::DeadTasksLockTimeout, /read tasks from/) + ensure + dead_tasks_storage.instance_variable_set(:@locked, false) + end + + it "keeps overlapping and Fiber-interleaved traversals independent" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + first = backend.enum_for(:each_dead_task) + second = backend.enum_for(:each_dead_task) + first_fiber = Fiber.new do + ids = [first.next[:id]] + Fiber.yield(ids.last) + ids << first.next[:id] + first.next + rescue StopIteration + ids + end + second_fiber = Fiber.new do + ids = [second.next[:id]] + Fiber.yield(ids.last) + ids << second.next[:id] + second.next + rescue StopIteration + ids + end + + expect(first_fiber.resume).to eq("1700000001-1-1") + expect(second_fiber.resume).to eq("1700000001-1-1") + expect(first_fiber.resume).to eq(["1700000001-1-1", "1700000002-1-2"]) + expect(second_fiber.resume).to eq(["1700000001-1-1", "1700000002-1-2"]) + end + + it "keeps a paused traversal usable when another traversal raises" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + paused = backend.enum_for(:each_dead_task) + + expect(paused.next[:id]).to eq("1700000001-1-1") + expect { backend.each_dead_task { raise "application failure" } }.to raise_error("application failure") + expect(finish_external_traversal(paused).map { |record| record[:id] }).to eq(["1700000002-1-2"]) + end + + it "establishes separate snapshots on each traversal's first advancement" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + first = backend.enum_for(:each_dead_task) + second = backend.enum_for(:each_dead_task) + + expect(first.next[:id]).to eq("1700000001-1-1") + add_dead_task("1700000003-1-3") + expect(finish_external_traversal(second).map { |record| record[:id] }).to eq( + ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"] + ) + expect(finish_external_traversal(first).map { |record| record[:id] }).to eq(["1700000002-1-2"]) + end + + it "keeps nested traversals independent" do + add_dead_task("1700000001-1-1") + add_dead_task("1700000002-1-2") + outer_ids = [] + nested_ids = nil + + backend.each_dead_task do |record| + outer_ids << record[:id] + nested_ids ||= each_dead_task.map { |nested| nested[:id] } + end + + expect(outer_ids).to eq(["1700000001-1-1", "1700000002-1-2"]) + expect(nested_ids).to eq(outer_ids) + end + end + + describe "#find_dead_task" do + let(:lookup_id) { "1700000001-1-1" } + + def append_lookup_record(outer_id, record) + entry = dead_tasks_storage.send(:build_entry, outer_id, record) + dead_tasks_path.open("ab") { |storage| storage.write(entry) } + end + + def append_lookup_payload(outer_id, serialized_record) + payload = "dead_task:#{outer_id}:#{serialized_record}" + crc = Zlib.crc32(payload).to_s(16).rjust(8, "0") + dead_tasks_path.open("ab") { |storage| storage.write("#{crc}:#{payload}\n") } + end + + def store_lookup_record(mode) + add_dead_task(lookup_id) + original = stored_records.last + + case mode + when :existing + original + when :duplicate + original.merge(task_class: "NewestTask", failed_at: original[:failed_at] + 1).tap do |record| + append_lookup_record(lookup_id, record) + end + when :frame_invalid_duplicate + dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:#{lookup_id}:secret\n") } + original + when :selected_winner_failure + append_lookup_payload(lookup_id, "not-a-dumped-string") + [RuntimeError, /dumped string/] + when :schema_incompatible_duplicate + original.merge(attempts: "invalid").tap do |record| + append_lookup_record(lookup_id, record) + end + end + end + + include_examples "a dead-task exact-lookup backend", empty: false + + it "silently skips malformed physical records without disclosing their payload" do + add_dead_task(lookup_id) + valid = stored_records.last + dead_tasks_path.open("ab") { |storage| storage.write("secret without a valid frame\n") } + + expect { expect(backend.find_dead_task(lookup_id)).to eq(valid) }.not_to output.to_stdout + expect { backend.find_dead_task(lookup_id) }.not_to output.to_stderr + end + + it "returns a frame-valid schema-incompatible match and excludes an incomplete tail" do + backend + record = { + id: lookup_id, task_class: "Task", attempts: "invalid", + enqueued_at: 1, failed_at: 2, exception_class: "RuntimeError", + exception_message: "secret", backtrace: [], context: Marshal.dump(["Task", [], {}]) + } + append_lookup_record(lookup_id, record) + dead_tasks_path.open("ab") { |storage| storage.write("incomplete-secret-tail") } + bytes = dead_tasks_path.binread + + expect(backend.find_dead_task(lookup_id)).to eq(record) + expect(dead_tasks_path.binread).to eq(bytes) + end + + it "completes the forward frame scan before decoding the newest match" do + add_dead_task(lookup_id) + add_dead_task("1700000002-1-2") + newest = stored_records.first.merge(exception_message: "newest") + append_lookup_record(lookup_id, newest) + add_dead_task("1700000003-1-3") + scanned_ids = [] + allow(Marshal).to receive(:load).and_call_original + allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |original, entry| + expect(Marshal).not_to have_received(:load) + original.call(entry).tap { |framed_record| scanned_ids << framed_record&.first } + end + + expect(backend.find_dead_task(lookup_id)[:exception_message]).to eq("newest") + expect(scanned_ids).to eq([lookup_id, "1700000002-1-2", lookup_id, "1700000003-1-3"]) + expect(Marshal).to have_received(:load).once + end + + it "uses the outer framed id when the decoded payload id differs" do + add_dead_task(lookup_id) + append_lookup_record(lookup_id, stored_records.last.merge(id: "different-id")) + + expect(backend.find_dead_task(lookup_id)[:id]).to eq(lookup_id) + end + + it "finds records with incompatible opaque contexts without loading them" do + add_dead_task(lookup_id) + record = stored_records.last.merge(context: "not Marshal") + append_lookup_record("1700000002-1-2", record.merge(id: "1700000002-1-2")) + allow(Marshal).to receive(:load).and_call_original + + found = backend.find_dead_task("1700000002-1-2") + + expect(found[:id]).to eq("1700000002-1-2") + expect(Marshal).to have_received(:load).once + end + + it "uses an independent snapshot without disturbing a paused enumeration" do + add_dead_task(lookup_id) + add_dead_task("1700000002-1-2") + dead_tasks_path.open("ab") { |storage| storage.write("malformed-secret\n") } + iterator = backend.enum_for(:each_dead_task) + + expect(iterator.next[:id]).to eq(lookup_id) + expect(backend.find_dead_task("1700000002-1-2")[:id]).to eq("1700000002-1-2") + expect(iterator.next[:id]).to eq("1700000002-1-2") + expect { iterator.next }.to raise_error(StopIteration) + end + + it "uses the complete-record boundary captured when lookup starts" do + add_dead_task(lookup_id) + original = stored_records.last + appended = false + newest = original.merge(exception_message: "added during lookup") + allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |method, entry| + method.call(entry).tap do + unless appended + appended = true + append_lookup_record(lookup_id, newest) + end + end + end + + expect(backend.find_dead_task(lookup_id)).to eq(original) + expect(backend.find_dead_task(lookup_id)).to eq(newest) + end + + it "propagates snapshot lock timeouts unchanged" do + stub_const("#{described_class}::DeadTasksStorage::LOCK_MAX_ATTEMPTS", 1) + dead_tasks_storage.instance_variable_set(:@locked, true) + + expect { + backend.find_dead_task(lookup_id) + }.to raise_error(Rage::Deferred::DeadTasksLockTimeout, /read tasks from/) + ensure + dead_tasks_storage.instance_variable_set(:@locked, false) + end + + it "propagates open failures unchanged" do + backend + allow(File).to receive(:open).and_call_original + allow(File).to receive(:open).with(dead_tasks_path, File::RDONLY | File::BINARY).and_raise(Errno::EACCES) + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EACCES) + end + + it "propagates forward-scan seek failures and closes the snapshot" do + add_dead_task(lookup_id) + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:seek).and_raise(Errno::EIO) + operation.call(storage, snapshot_end) + end + end + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO) + expect(descriptor).to be_closed + end + + it "propagates forward-scan read failures and closes the snapshot" do + add_dead_task(lookup_id) + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:gets).and_raise(Errno::EIO) + operation.call(storage, snapshot_end) + end + end + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO) + expect(descriptor).to be_closed + end + + it "does not let cleanup replace an active lookup failure" do + add_dead_task(lookup_id) + descriptor = nil + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + descriptor = storage + allow(descriptor).to receive(:seek).and_raise(Errno::EACCES) + allow(descriptor).to receive(:close).and_wrap_original do |close| + close.call + raise Errno::EIO + end + operation.call(storage, snapshot_end) + end + end + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EACCES) + expect(descriptor).to be_closed + end + + it "propagates a descriptor cleanup error when it is the only failure" do + add_dead_task(lookup_id) + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + allow(storage).to receive(:close).and_wrap_original do |close| + close.call + raise Errno::EIO + end + operation.call(storage, snapshot_end) + end + end + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO) + end + + it "propagates an operational failure after corruption and a provisional match" do + backend + dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:#{lookup_id}:secret\n") } + add_dead_task(lookup_id) + add_dead_task("1700000002-1-2") + allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation| + original.call do |storage, snapshot_end| + reads = 0 + allow(storage).to receive(:gets).and_wrap_original do |gets| + reads += 1 + raise Errno::EIO if reads == 3 + + gets.call + end + operation.call(storage, snapshot_end) + end + end + + expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO) + end + end + describe "#remove_dead_tasks" do it "removes the given dead tasks and keeps the others" do add_dead_task("1-1-1") add_dead_task("2-2-2") expect(backend.remove_dead_tasks("1-1-1")).to eq(1) - expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["2-2-2"]) + expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["2-2-2"]) expect(tmp_storage_path).not_to exist end @@ -543,7 +1263,7 @@ def stored_records add_dead_task("1-1-1") expect(backend.remove_dead_tasks("nope")).to eq(0) - expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["1-1-1"]) + expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1"]) expect(tmp_storage_path).not_to exist end diff --git a/spec/deferred/backends/nil_spec.rb b/spec/deferred/backends/nil_spec.rb new file mode 100644 index 00000000..07cb3919 --- /dev/null +++ b/spec/deferred/backends/nil_spec.rb @@ -0,0 +1,13 @@ +# frozen_string_literal: true + +RSpec.describe Rage::Deferred::Backends::Nil do + subject(:backend) { described_class.new } + + it "retains traversal and removal behavior" do + expect(backend).to respond_to(:each_dead_task, :find_dead_task, :remove_dead_tasks) + expect(backend.enum_for(:each_dead_task).to_a).to eq([]) + expect(backend.remove_dead_tasks("missing-id")).to eq(0) + end + + include_examples "a dead-task exact-lookup backend", empty: true +end diff --git a/spec/deferred/dead_tasks_spec.rb b/spec/deferred/dead_tasks_spec.rb new file mode 100644 index 00000000..907d63bc --- /dev/null +++ b/spec/deferred/dead_tasks_spec.rb @@ -0,0 +1,389 @@ +# frozen_string_literal: true + +RSpec.describe Rage::Deferred::DeadTasks do + let(:records) do + [ + { + id: "1700000001-1-1", task_class: "OlderTask", attempts: 2, + enqueued_at: 1_700_000_001, failed_at: 1_700_000_011, + exception_class: "ArgumentError", exception_message: "bad argument", + backtrace: [], context: Marshal.dump(["OlderTask", [], {}]) + }, + { + id: "1700000002-1-2", task_class: "MissingTask", attempts: 4, + enqueued_at: 1_700_000_002, failed_at: 1_700_000_012, + exception_class: "RuntimeError", exception_message: "secret failure", + backtrace: ["app/task.rb:1"], context: Marshal.dump(["MissingTask", [], {}]) + } + ] + end + let(:backend) do + stored_records = records + Class.new do + attr_reader :traversal_count, :lookup_count + + define_method(:initialize) do + @traversal_count = 0 + @lookup_count = 0 + end + define_method(:each_dead_task) do |&block| + @traversal_count += 1 + stored_records.each(&block) + self + end + define_method(:find_dead_task) do |id| + @lookup_count += 1 + stored_records.reverse_each.find { |record| record[:id] == id } + end + end.new + end + subject(:dead_tasks) { described_class.new(backend) } + + describe "#each" do + it "returns distinct standard Enumerators without starting backend traversal" do + expect(backend).not_to receive(:each_dead_task) + + first = dead_tasks.each + second = dead_tasks.each + + expect(first).to be_a(Enumerator) + expect(second).to be_a(Enumerator) + expect(first).not_to equal(second) + expect(first).not_to respond_to(:close) + end + + it "keeps both the public and private backend traversal APIs argument-free" do + expect(dead_tasks.method(:each).parameters).to eq([]) + expect(backend.method(:each_dead_task).parameters).to eq([[:block, :block]]) + expect { dead_tasks.each(limit: 1) }.to raise_error(ArgumentError) + + dead_tasks.each {} + + expect(backend.traversal_count).to eq(1) + end + + it "yields passive defensive summaries without resolving task classes or contexts" do + expect(Object).not_to receive(:const_get) + expect(Marshal).not_to receive(:load) + + entries = dead_tasks.to_a + + expect(entries.map(&:id)).to eq(["1700000001-1-1", "1700000002-1-2"]) + expect(entries.last.task_class).to eq("MissingTask") + expect(entries.last.attempts).to eq(4) + expect(entries.last.enqueued_at).to eq(Time.at(1_700_000_002)) + expect(entries.last.failed_at).to eq(Time.at(1_700_000_012)) + expect(entries.last.id).to be_frozen + expect(entries.last.task_class).to be_frozen + expect(entries.last.enqueued_at.utc).to equal(entries.last.enqueued_at) + expect(entries.last.failed_at.gmtime).to equal(entries.last.failed_at) + entries.last.enqueued_at.localtime("+05:00") + entries.last.failed_at.localtime("-04:00") + fresh_entry = dead_tasks.to_a.last + + expect(entries.last.enqueued_at.utc_offset).to eq(5 * 60 * 60) + expect(entries.last.failed_at.utc_offset).to eq(-4 * 60 * 60) + expect(fresh_entry.enqueued_at).to eq(Time.at(1_700_000_002)) + expect(fresh_entry.failed_at).to eq(Time.at(1_700_000_012)) + expect(fresh_entry.enqueued_at).not_to equal(entries.last.enqueued_at) + expect(fresh_entry.failed_at).not_to equal(entries.last.failed_at) + expect(records.last.values_at(:enqueued_at, :failed_at)).to eq([1_700_000_002, 1_700_000_012]) + expect(entries.last).to respond_to(:args, :kwargs, :exception_class, :exception_message, :backtrace) + end + + it "retains standard Enumerable find and detect behavior" do + fallback = -> { :missing } + unused_fallback = -> { raise "must not run" } + + expect(dead_tasks.find { |entry| entry.attempts == 2 }.task_class).to eq("OlderTask") + expect(dead_tasks.find(unused_fallback) { |entry| entry.attempts == 2 }.task_class).to eq("OlderTask") + expect(dead_tasks.detect(fallback) { |entry| entry.attempts == 99 }).to eq(:missing) + expect(dead_tasks.find).to be_a(Enumerator) + expect(dead_tasks.detect).to be_a(Enumerator) + end + + it "uses independent traversals for separate Enumerable operations" do + dead_tasks.first + dead_tasks.count + + expect(backend.traversal_count).to eq(2) + expect(dead_tasks.instance_variables).to eq([:@backend]) + expect(dead_tasks.instance_variable_get(:@backend)).to equal(backend) + end + end + + describe "#find_by_id" do + it "validates an exact String before backend access without coercion" do + expect(backend).not_to receive(:find_dead_task) + + expect { dead_tasks.find_by_id(:"1700000001-1-1") }.to raise_error(TypeError, /String/) + expect { dead_tasks.find_by_id(1_700_000_001) }.to raise_error(TypeError, /String/) + expect(backend.lookup_count).to eq(0) + end + + it "delegates polymorphically and wraps a matching record" do + entry = dead_tasks.find_by_id("1700000002-1-2") + + expect(entry).to be_a(Rage::Deferred::DeadTask) + expect(entry.id).to eq("1700000002-1-2") + expect(backend.lookup_count).to eq(1) + expect(dead_tasks.find_by_id("missing")).to be_nil + end + + it "uses the newest backend record without schema validation or fallback" do + id = "schema-drift" + records << records.first.merge(id:, attempts: 1, context: Marshal.dump(["Task", ["older"], {}])) + records << records.first.merge(id:, task_class: nil, attempts: "unvalidated", context: "not Marshal") + + entry = dead_tasks.find_by_id(id) + + expect(entry.task_class).to be_nil + expect(entry.attempts).to eq("unvalidated") + expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) + end + + it "propagates natural entry-construction failures from the selected backend record" do + invalid_records = [ + [[], TypeError], + [records.first.merge(backtrace: "not an array"), NoMethodError] + ] + + invalid_records.each do |record, error_class| + selected_backend = instance_double("Rage::Deferred::Backend", find_dead_task: record) + collection = described_class.new(selected_backend) + + expect { collection.find_by_id("selected") }.to raise_error(error_class) + end + end + + it "exposes frozen exception metadata without decoding context or resolving the task class" do + expect(Object).not_to receive(:const_get) + expect(Marshal).not_to receive(:load) + + entry = dead_tasks.find_by_id("1700000002-1-2") + + expect(entry.exception_class).to eq("RuntimeError") + expect(entry.exception_message).to eq("secret failure") + expect(entry.backtrace).to eq(["app/task.rb:1"]) + expect(entry.exception_class).to be_frozen + expect(entry.exception_message).to be_frozen + expect(entry.backtrace).to be_frozen + expect(entry.backtrace.first).to be_frozen + expect { entry.backtrace << "changed" }.to raise_error(FrozenError) + end + + it "lazily decodes empty, positional, keyword, and mixed arguments with stable frozen identities" do + contexts = [ + ["Task", nil, nil], + ["Task", [["nested"]], nil], + ["Task", nil, { key: { nested: "value" } }], + ["Task", [1], { key: 2 }] + ] + load_count = 0 + allow(Marshal).to receive(:load).and_wrap_original do |original, *args, **kwargs| + load_count += 1 + original.call(*args, **kwargs) + end + + contexts.each_with_index do |context, index| + record = records.first.merge(id: "context-#{index}", context: Marshal.dump(context)) + records << record + entry = dead_tasks.find_by_id(record[:id]) + + expect(entry.args).to eq(context[1] || []) + expect(entry.kwargs).to eq(context[2] || {}) + expect(entry.args).to equal(entry.args) + expect(entry.kwargs).to equal(entry.kwargs) + expect(entry.args).to be_frozen + expect(entry.kwargs).to be_frozen + end + + expect(load_count).to eq(contexts.length) + expect(dead_tasks.find_by_id("context-1").args.dig(0)).to eq(["nested"]) + expect(dead_tasks.find_by_id("context-2").kwargs.dig(:key, :nested)).to eq("value") + end + + it "uses the deferred context accessors to extract positional and keyword arguments" do + context = ["Task", [1], { key: 2 }] + records << records.first.merge(id: "context-accessors", context: Marshal.dump(context)) + entry = dead_tasks.find_by_id("context-accessors") + + expect(Marshal).to receive(:load).with(instance_of(String), freeze: true).and_call_original + expect(Rage::Deferred::Context).to receive(:get_args).with(instance_of(Array)).and_call_original + expect(Rage::Deferred::Context).to receive(:get_kwargs).with(instance_of(Array)).and_call_original + + expect(entry.args).to eq([1]) + expect(entry.kwargs).to eq({ key: 2 }) + end + + it "normalizes missing argument slots through the context accessors" do + records << records.first.merge(id: "missing-slots", context: Marshal.dump(["Task"])) + entry = dead_tasks.find_by_id("missing-slots") + + expect(entry.args).to eq([]) + expect(entry.kwargs).to eq({}) + expect(entry.args).to be_frozen + expect(entry.kwargs).to be_frozen + expect(entry.args).to equal(entry.args) + expect(entry.kwargs).to equal(entry.kwargs) + end + + it "freezes nested argument graphs defensively" do + shared = ["secret"] + shared << shared + records << records.first.merge( + id: "nested", context: Marshal.dump(["Task", [shared], { shared: }]) + ) + entry = dead_tasks.find_by_id("nested") + + expect(entry.args.first).to equal(entry.kwargs[:shared]) + expect(entry.args.first).to equal(entry.args.first.last) + expect(entry.args.first).to be_frozen + expect(entry.args.first.first).to be_frozen + expect { entry.args.first << "changed" }.to raise_error(FrozenError) + expect { entry.kwargs[:shared].first.replace("changed") }.to raise_error(FrozenError) + end + + it "raises the dedicated error with id and cause, without caching a failed decode" do + records << records.first.merge(id: "broken", context: "not Marshal") + entry = dead_tasks.find_by_id("broken") + + 2.times do + expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error| + expect(error.message).to include("broken") + expect(error.cause).to be_a(TypeError) + } + end + expect(entry.id).to eq("broken") + expect(entry.exception_class).to eq("ArgumentError") + expect(dead_tasks.find_by_id("broken")).not_to be_nil + end + + it "translates context accessor failures without caching a partial extraction" do + records << records.first.merge(id: "accessor-failure", context: Marshal.dump(["Task", [1], { key: 2 }])) + entry = dead_tasks.find_by_id("accessor-failure") + args_calls = 0 + kwargs_calls = 0 + + allow(Rage::Deferred::Context).to receive(:get_args).and_wrap_original do |method, context| + args_calls += 1 + method.call(context) + end + allow(Rage::Deferred::Context).to receive(:get_kwargs).and_wrap_original do |method, context| + kwargs_calls += 1 + raise NoMethodError, "context accessor failed" if kwargs_calls == 1 + + method.call(context) + end + + expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error| + expect(error.message).to include("accessor-failure") + expect(error.cause).to be_a(NoMethodError) + } + expect(entry.kwargs).to eq({ key: 2 }) + expect(entry.args).to eq([1]) + expect(args_calls).to eq(2) + expect(kwargs_calls).to eq(2) + end + + it "keeps missing referenced constants recoverable through metadata" do + stub_const("TemporaryDeadTaskArgument", Class.new) + dumped = Marshal.dump(["Task", [TemporaryDeadTaskArgument.new], {}]) + hide_const("TemporaryDeadTaskArgument") + records << records.first.merge(id: "missing-constant", context: dumped) + entry = dead_tasks.find_by_id("missing-constant") + + expect(entry.task_class).to eq("OlderTask") + expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error| + expect(error.cause).to be_a(ArgumentError).or be_a(NameError) + } + end + + it "retries decoding after a missing referenced constant is restored" do + stub_const("RestorableDeadTaskArgument", Class.new) + dumped = Marshal.dump(["Task", [RestorableDeadTaskArgument.new], {}]) + hide_const("RestorableDeadTaskArgument") + records << records.first.merge(id: "restorable-constant", context: dumped) + entry = dead_tasks.find_by_id("restorable-constant") + + expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) + + stub_const("RestorableDeadTaskArgument", Class.new) + expect(entry.args.first).to be_a(RestorableDeadTaskArgument) + end + end +end + +RSpec.describe Rage::Deferred, ".dead_tasks" do + before do + @memoized_values = {} + %i[@__backend @__dead_tasks @__queue].each do |name| + if described_class.instance_variable_defined?(name) + @memoized_values[name] = described_class.instance_variable_get(name) + described_class.remove_instance_variable(name) + end + end + end + + after do + %i[@__backend @__dead_tasks @__queue].each do |name| + described_class.remove_instance_variable(name) if described_class.instance_variable_defined?(name) + described_class.instance_variable_set(name, @memoized_values[name]) if @memoized_values.key?(name) + end + end + + let(:backend) { instance_double("Rage::Deferred::Backend") } + + before do + allow(Rage.config.deferred).to receive(:backend).and_return(backend) + end + + it "resolves the configured backend once and passes it directly to one memoized collection" do + expect(Rage::Deferred::DeadTasks).to receive(:new).with(backend).once.and_call_original + expect(backend).not_to receive(:each_dead_task) + + first = described_class.dead_tasks + + expect(described_class.dead_tasks).to equal(first) + expect(first.instance_variable_get(:@backend)).to equal(backend) + expect(Rage.config.deferred).to have_received(:backend).once + end + + it "does not start traversal when creating unadvanced Enumerators" do + expect(backend).not_to receive(:each_dead_task) + + first = described_class.dead_tasks.each + second = described_class.dead_tasks.each + + expect(first).to be_a(Enumerator) + expect(second).not_to equal(first) + end + + it "shares the backend when the queue initializes it first" do + queue = described_class.__queue + collection = described_class.dead_tasks + + expect(queue.instance_variable_get(:@backend)).to equal(backend) + expect(collection.instance_variable_get(:@backend)).to equal(backend) + expect(Rage.config.deferred).to have_received(:backend).once + end + + it "shares the backend when the collection initializes it first" do + collection = described_class.dead_tasks + queue = described_class.__queue + + expect(collection.instance_variable_get(:@backend)).to equal(backend) + expect(queue.instance_variable_get(:@backend)).to equal(backend) + expect(Rage.config.deferred).to have_received(:backend).once + end +end + +RSpec.describe Rage::Deferred::Backends::Nil, "dead-task traversal" do + it "is empty without creating persistence or background activity" do + backend = described_class.new + collection = Rage::Deferred::DeadTasks.new(backend) + + expect(backend.method(:each_dead_task).parameters).to eq([]) + expect(collection.to_a).to eq([]) + end +end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index c6de6aca..913118f0 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -9,6 +9,7 @@ require_relative "support/contexts/mocked_classes" require_relative "support/contexts/mocked_rage_routes" require_relative "support/custom_matchers" +require_relative "support/shared_examples/dead_task_exact_lookup" RSpec.configure do |config| # Uncomment the line below to enable focused mode diff --git a/spec/support/shared_examples/dead_task_exact_lookup.rb b/spec/support/shared_examples/dead_task_exact_lookup.rb new file mode 100644 index 00000000..cd0f8a09 --- /dev/null +++ b/spec/support/shared_examples/dead_task_exact_lookup.rb @@ -0,0 +1,50 @@ +# frozen_string_literal: true + +RSpec.shared_examples "a dead-task exact-lookup backend" do |empty:| + it "returns nil for a missing exact id" do + expect(backend.find_dead_task("missing-id")).to be_nil + end + + if empty + it "remains empty and side-effect free" do + state = backend.instance_variables.to_h { |name| [name, backend.instance_variable_get(name)] } + + expect(backend.find_dead_task("1700000001-1-1")).to be_nil + expect(backend.instance_variables.to_h { |name| [name, backend.instance_variable_get(name)] }).to eq(state) + end + else + it "returns an existing record with the shared public shape" do + expected = store_lookup_record(:existing) + + expect(backend.find_dead_task(lookup_id)).to eq(expected) + expect(expected).to include( + :id, :task_class, :attempts, :enqueued_at, :failed_at, + :exception_class, :exception_message, :backtrace, :context + ) + end + + it "returns the newest frame-valid duplicate" do + expected = store_lookup_record(:duplicate) + + expect(backend.find_dead_task(lookup_id)).to eq(expected) + end + + it "retains the older frame-valid duplicate when the newer match is frame-invalid" do + expected = store_lookup_record(:frame_invalid_duplicate) + + expect(backend.find_dead_task(lookup_id)).to eq(expected) + end + + it "propagates a selected winner failure without falling back to an older duplicate" do + error_class, message = store_lookup_record(:selected_winner_failure) + + expect { backend.find_dead_task(lookup_id) }.to raise_error(error_class, message) + end + + it "returns a schema-incompatible newer frame without falling back" do + expected = store_lookup_record(:schema_incompatible_duplicate) + + expect(backend.find_dead_task(lookup_id)).to eq(expected) + end + end +end From 102f12f237702b071d0d8f2a4583e0aaad40e2c3 Mon Sep 17 00:00:00 2001 From: Alex Rohachev Date: Mon, 5 Oct 2026 19:58:42 +0000 Subject: [PATCH 2/4] [Deferred] Add dead tasks changelog entry Co-authored-by: Serhii Sadovskyi --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 133a0e01..f8761d5e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### Added +- [Deferred] Add `Rage::Deferred.dead_tasks` for listing and inspecting tasks that exhausted their retries. - Add low-level `PubSub` API (#399). - [Cable] Serialize WS messages per connection (#400). - Update fiber scheduler to support Ruby 4.1 (#397). From 8521e866bdfb9217fb9c8b23af915537bfffdc00 Mon Sep 17 00:00:00 2001 From: Alex Rohachev Date: Tue, 6 Oct 2026 06:43:04 +0000 Subject: [PATCH 3/4] Improve dead task documentation Co-authored-by: Serhii Sadovskyi --- lib/rage/deferred/backends/disk.rb | 92 +++++++++++++++++------------- lib/rage/deferred/backends/nil.rb | 7 --- lib/rage/deferred/dead_task.rb | 51 +++++++---------- lib/rage/deferred/dead_tasks.rb | 17 ++---- lib/rage/deferred/deferred.rb | 20 ++++--- 5 files changed, 92 insertions(+), 95 deletions(-) diff --git a/lib/rage/deferred/backends/disk.rb b/lib/rage/deferred/backends/disk.rb index 9c328b9d..7b4952bd 100644 --- a/lib/rage/deferred/backends/disk.rb +++ b/lib/rage/deferred/backends/disk.rb @@ -10,9 +10,10 @@ # * `remove_task` - called when a task has to be removed from the storage; # * `pending_tasks` - the method should iterate over the underlying storage and return a list of tasks to replay; # * `add_dead_task` - called when a task has exhausted its retries or aborted them; -# * `each_dead_task` - calls the block once for each logical dead task in a fixed snapshot, -# ordered by the physical positions of the selected winning records; -# * `find_dead_task` - returns the newest frame-valid dead-task record with the exact outer ID; +# * `each_dead_task` - yields one record for each task ID that exists when traversal starts; +# if multiple records have the same task ID, it yields the newest record; +# records added during traversal are not included; +# * `find_dead_task` - returns the newest dead-task record with the exact task ID; # * `remove_dead_tasks` - permanently delete dead tasks; # class Rage::Deferred::Backends::Disk @@ -54,19 +55,20 @@ def add_dead_task(task_id, context, exception, task_class:, attempts:) @dead_tasks_storage.add(task_id, context, exception, task_class:, attempts:) end - # Call the block once for each logical dead task in a fixed snapshot. - # Selected winners are ordered by their physical positions from oldest to newest. - # Changes to the live storage during iteration do not change the selected records. - # @yieldparam record [Hash] a decoded selected dead-task record + # Yield one record for each task ID that exists when traversal starts. + # If multiple records have the same task ID, yield the newest record. + # Do not include records that Rage adds during traversal. + # Yield records from oldest to newest. + # @yieldparam record [Hash] a dead-task record # @return [Rage::Deferred::Backends::Disk::DeadTasksStorage] # @private def each_dead_task(&block) @dead_tasks_storage.each(&block) end - # Return the newest frame-valid dead-task record with the exact outer ID. - # @param id [String] the persisted task ID - # @return [Hash, nil] the decoded record, or nil if there is no matching record + # Return the newest dead-task record with the exact task ID. + # @param id [String] the task ID + # @return [Hash, nil] the matching record, or nil if there is no matching record # @private def find_dead_task(id) @dead_tasks_storage.find_by_id(id) @@ -425,17 +427,19 @@ def add(task_id, context, exception, task_class:, attempts:) end end - # Call the block once for each selected dead-task record, oldest winning position first. + # Yield each selected dead-task record from oldest to newest. # - # The method opens the current file while it holds the lock. - # It sets the read limit after the last complete record and then releases the lock. - # Records added after this point are not part of the traversal. - # After releasing the lock, it scans complete records sequentially forwards and selects - # the newest frame-valid record for each authoritative outer task ID. Repeated IDs are - # ordered at their winning records' physical positions. It then seeks to, decodes, and - # yields one selected payload at a time, without decoding the next winner before control - # returns from the current yield. - # Each call has separate traversal state. + # The method locks the store and opens the current file. + # It sets a read boundary after the last complete record. + # It then releases the lock. + # A record that Rage adds after this step is not part of the traversal. + # + # The method scans each complete record before it yields the first record. + # For each task ID, it selects the newest record that has a valid frame. + # + # The method decodes one selected record immediately before it yields that record. + # It does not decode the next record while the caller processes the current record. + # Each call uses independent traversal state. # # @yieldparam record [Hash] a decoded dead-task record # @return [self] @@ -452,9 +456,11 @@ def each end end - # Return the newest frame-valid record with the exact authoritative outer ID. - # Scan the fixed complete-record snapshot forwards without decoding candidate payloads. - # Decode only the newest matching payload after scanning reaches the snapshot boundary. + # Return the newest record with the exact task ID. + # + # Scan all complete records in the fixed snapshot. + # Do not decode a candidate record during the scan. + # After the scan, decode only the newest matching record. # @param id [String] the persisted task ID # @return [Hash, nil] the decoded record, or nil if there is no matching record # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked @@ -523,9 +529,10 @@ def remove(ids) private - # Index the newest frame-valid serialized-payload location for each authoritative outer ID. - # Deleting and reinserting a repeated ID moves it to its winning record's physical position, - # so Hash insertion order visits the completed index from oldest winner to newest winner. + # Index the newest valid payload location for each task ID. + # If the index contains the duplicate task ID, the method removes the ID before it adds the newer record. + # This operation moves the ID to the position of its newest record. + # Hash insertion order then gives the required order from oldest to newest. # @param storage [File] the descriptor for the fixed snapshot # @param snapshot_end [Integer] the byte boundary after its last complete record # @return [Hash{String => Array<(Integer, Integer)>}] ordered serialized-payload locations @@ -539,11 +546,12 @@ def records_index(storage, snapshot_end) records_index end - # Scan complete physical records forwards through a fixed snapshot boundary. - # Silently skip malformed frames and invalid CRCs. Do not decode or inspect serialized payloads. + # Scan each complete record up to the fixed snapshot boundary. + # Silently skip a malformed frame or an invalid CRC. + # Do not decode or inspect a serialized payload. # @param storage [File] the descriptor for the fixed snapshot # @param snapshot_end [Integer] the byte boundary after its last complete record - # @yieldparam task_id [String] the authoritative outer task ID + # @yieldparam task_id [String] the task ID from the storage frame # @yieldparam offset [Integer] the serialized payload's byte offset # @yieldparam length [Integer] the serialized payload's byte length # @return [void] @@ -568,10 +576,12 @@ def each_framed_record(storage, snapshot_end) end end - # Open the live file and capture its last complete-record boundary under the permanent lock. - # Release the lock before scanning, decoding, or yielding. The retained descriptor continues - # to identify the same snapshot file if the live path is replaced while the block runs. - # Close the descriptor on unwind; a close error does not replace an active exception. + # Lock the store, open the live file, and find the last complete record. + # Release the lock before the block scans, decodes, or yields a record. + # Keep the file descriptor open while the block runs. + # The descriptor continues to identify the same file if another operation replaces the live path. + # Close the descriptor when the block exits. + # If the block and `File#close` both raise exceptions, the method re-raises the block's exception. # @yieldparam storage [File] the descriptor for the fixed snapshot # @yieldparam snapshot_end [Integer] the fixed end position of the last complete record # @yieldreturn [Object] the result from the read operation @@ -599,9 +609,12 @@ def with_snapshot end end - # Locate the byte boundary immediately after the last newline-terminated record. - # If the file has an incomplete tail, scan backwards in bounded chunks without parsing records. - # Do not modify the file; return zero when it contains no complete record. + # Return the byte position after the last complete record. + # A complete record ends with a newline. + # If the file has an incomplete tail, scan backwards in fixed-size chunks. + # Do not parse a record during this scan. + # Do not change the file. + # Return zero if the file has no complete record. # @param storage [File] an open dead-task file # @return [Integer] the byte boundary after the last complete record def complete_record_end(storage) @@ -628,9 +641,10 @@ def complete_record_end(storage) 0 end - # Check only the outer storage frame and return its authoritative ID and serialized payload. - # Return nil for a malformed frame, invalid CRC, unexpected operation, or empty outer ID. - # Do not decode the payload or validate its shape or any ID stored inside it. + # Validate the storage frame. + # Return its task ID and serialized payload. + # Return `nil` for a malformed frame, invalid CRC, unsupported operation, or empty task ID. + # Do not decode the payload. # @param entry [String] one complete physical record without the newline # @return [Array<(String, String)>, nil] def framed_record(entry) diff --git a/lib/rage/deferred/backends/nil.rb b/lib/rage/deferred/backends/nil.rb index ab75a932..83d45e7a 100644 --- a/lib/rage/deferred/backends/nil.rb +++ b/lib/rage/deferred/backends/nil.rb @@ -17,17 +17,10 @@ def pending_tasks def add_dead_task(_, _, _, **) end - # Yield no dead-task records. - # @return [self] - # @private def each_dead_task self end - # Return no dead task for an exact ID lookup. - # @param _ [String] the persisted task ID - # @return [nil] - # @private def find_dead_task(_) end diff --git a/lib/rage/deferred/dead_task.rb b/lib/rage/deferred/dead_task.rb index 8aa43d7c..1c997748 100644 --- a/lib/rage/deferred/dead_task.rb +++ b/lib/rage/deferred/dead_task.rb @@ -1,36 +1,34 @@ # frozen_string_literal: true ## -# A read-only view of a deferred task that used all its retries or stopped retrying. +# Provides information about one dead task. # -# Values come from trusted local dead-task storage. -# Reading metadata does not load the task class or deserialize the stored execution context. -# {#args} and {#kwargs} load the context only when you call them. -# The context is trusted local Marshal data. -# Argument data can contain credentials or personal data. Protect this data from unauthorized access. +# A dead task is a deferred task that is no longer retried. +# Creating this object does not load the task class. +# The first call to {#args} or {#kwargs} deserializes the execution context. class Rage::Deferred::DeadTask - # @return [String] the task's persisted identifier + # @return [String] the ID that Rage assigned when it enqueued the task attr_reader :id - # @return [String] the stored task class name + # @return [String] the name of the task class attr_reader :task_class - # @return [Integer] the number of attempts to process the task + # @return [Integer] the number of processing attempts attr_reader :attempts - # @return [Time] the time encoded in the task's persisted identifier + # @return [Time] the time when Rage enqueued the task attr_reader :enqueued_at - # @return [Time] the time when the final attempt failed + # @return [Time] the time when the last attempt failed attr_reader :failed_at - # @return [String] the stored class name of the final exception + # @return [String] the class name of the last exception attr_reader :exception_class - # @return [String] the stored message of the final exception + # @return [String] the message from the last exception attr_reader :exception_message - # @return [Array] the frozen stored backtrace of the final exception + # @return [Array] the frozen backtrace from the last exception attr_reader :backtrace # @private @@ -49,14 +47,10 @@ def initialize(record) # Return the original positional arguments. # - # The first successful call decodes the trusted local Marshal context. - # The method freezes the decoded object graph and returns the same Array on later calls. - # Changes to the returned Array cannot change the stored replay input. - # A stored `nil` produces an empty Array. - # The method does not cache a decoding failure. - # You can try again after you restore the application constants. - # - # Argument data can contain credentials or personal data. Protect this data from unauthorized access. + # On the first call, the method deserializes the execution context. + # The method freezes all objects in the decoded data. + # After a successful call, the method returns the same Array on each later call. + # If the stored value is `nil`, the method returns a frozen empty Array. # # @return [Array] frozen positional arguments # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded @@ -67,15 +61,10 @@ def args # Return the original keyword arguments. # - # The first successful call decodes the trusted local Marshal context. - # The method freezes the decoded object graph and returns the same Hash on later calls. - # Changes to the returned Hash cannot change the stored replay input. - # A stored `nil` produces an empty Hash. - # This behavior keeps positional and keyword arguments separate on Ruby 3.3. - # The method does not cache a decoding failure. - # You can try again after you restore the application constants. - # - # Argument data can contain credentials or personal data. Protect this data from unauthorized access. + # On the first call, the method deserializes the execution context. + # The method freezes all objects in the decoded data. + # After a successful call, the method returns the same Hash on each later call. + # If the stored value is `nil`, the method returns a frozen empty Hash. # # @return [Hash] frozen keyword arguments # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded diff --git a/lib/rage/deferred/dead_tasks.rb b/lib/rage/deferred/dead_tasks.rb index f703c156..7ba38d47 100644 --- a/lib/rage/deferred/dead_tasks.rb +++ b/lib/rage/deferred/dead_tasks.rb @@ -1,13 +1,9 @@ # frozen_string_literal: true ## -# The collection of deferred tasks that used all their retries or stopped retrying. +# Provides access to deferred tasks that Rage no longer retries. # -# The first call to {Rage::Deferred.dead_tasks} creates a collection wrapper around the configured backend. -# Later calls return the same wrapper. -# The wrapper uses the same backend object as the deferred queue. -# The wrapper does not keep traversal state. -# Each traversal requests records from the configured backend. +# {Rage::Deferred.dead_tasks} returns this collection. class Rage::Deferred::DeadTasks include Enumerable @@ -16,12 +12,12 @@ def initialize(backend) @backend = backend end - # Call the block for each read-only dead-task summary. Start with the oldest task. + # Yield each dead task from oldest to newest. # - # Without a block, this method returns a standard lazy Ruby Enumerator. + # Without a block, the method returns a new Enumerator. # - # @yieldparam dead_task [Rage::Deferred::DeadTask] a read-only task summary - # @return [Enumerator, self] a new Enumerator without a block, otherwise this collection + # @yieldparam dead_task [Rage::Deferred::DeadTask] information about one dead task + # @return [Enumerator, self] a new Enumerator without a block, or this collection with a block def each return enum_for(__method__) unless block_given? @@ -35,7 +31,6 @@ def each # Return the dead task with the exact persisted ID. # # The ID must be a String. The method does not convert other values to a String. - # The method does not load the stored task class or deserialize its execution context. # # @param id [String] the exact persisted task ID # @return [Rage::Deferred::DeadTask, nil] the matching task, or nil diff --git a/lib/rage/deferred/deferred.rb b/lib/rage/deferred/deferred.rb index 5fe2a06f..642f5a59 100644 --- a/lib/rage/deferred/deferred.rb +++ b/lib/rage/deferred/deferred.rb @@ -51,12 +51,18 @@ def self.wrap(instance, delay: nil, delay_until: nil) Rage::Deferred::Proxy.new(instance, delay:, delay_until:) end - # Return the collection of tasks that used all their retries or stopped retrying. + # Return the collection of tasks that Rage no longer retries. # - # The method returns the same collection for each call. + # The method returns the same collection object on each call. # The collection uses the same backend object as the deferred queue. - # The first call can initialize the backend. - # Obtaining the collection does not open, scan, or decode dead-task records. + # + # @example List dead tasks + # Rage::Deferred.dead_tasks.each do |task| + # puts "#{task.id}: #{task.exception_message}" + # end + # + # @example Find one dead task + # task = Rage::Deferred.dead_tasks.find_by_id("1735689600-12345-1") # # @return [Rage::Deferred::DeadTasks] def self.dead_tasks @@ -112,14 +118,14 @@ module Backends class PushTimeout < StandardError end - # Raised when a dead-task store operation cannot get the lock after all retries. + # Raise this error when the dead-task store cannot be locked after all retries. class DeadTasksLockTimeout < StandardError end - # Raised when the method cannot decode the stored execution context of a dead task. + # Raise this error when a dead task's execution context cannot be deserialized. # # The error message identifies the dead task. - # `Exception#cause` contains the original Marshal error or context-layout error. + # `Exception#cause` contains the original deserialization error. class DeadTaskContextDeserializationError < StandardError end end From 305d7defd9f2f8f2e4ca5ae346b33fc7f35350f3 Mon Sep 17 00:00:00 2001 From: Serhii Sadovskyi Date: Thu, 8 Oct 2026 15:40:27 +0000 Subject: [PATCH 4/4] [Deferred] Skip dead tasks that cannot be deserialized `each` and `find_by_id` now rescue `Marshal.load` errors and print the task ID. `each` continues with the next record, and `find_by_id` returns `nil`. Co-authored-by: Alex Rohachev --- lib/rage/deferred/backends/disk.rb | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/lib/rage/deferred/backends/disk.rb b/lib/rage/deferred/backends/disk.rb index 7b4952bd..ba6dea6b 100644 --- a/lib/rage/deferred/backends/disk.rb +++ b/lib/rage/deferred/backends/disk.rb @@ -449,7 +449,16 @@ def each with_snapshot do |storage, snapshot_end| records_index(storage, snapshot_end).each do |task_id, (offset, length)| storage.seek(offset, IO::SEEK_SET) - yield Marshal.load(storage.read(length).undump).merge(id: task_id) + + begin + record = Marshal.load(storage.read(length).undump) + rescue ArgumentError, NameError, TypeError => e + puts "ERROR: Can't deserialize the dead-lettered task with id #{task_id}: (#{e.class}) #{e.message}" + next + end + + # yield outside the `rescue`, so that an error raised by the caller's block is not swallowed + yield record.merge(id: task_id) end self @@ -478,7 +487,15 @@ def find_by_id(id) offset, length = serialized_record_location storage.seek(offset, IO::SEEK_SET) - Marshal.load(storage.read(length).undump).merge(id: id) + + begin + record = Marshal.load(storage.read(length).undump) + rescue ArgumentError, NameError, TypeError => e + puts "ERROR: Can't deserialize the dead-lettered task with id #{id}: (#{e.class}) #{e.message}" + next + end + + record.merge(id: id) end end