Class: Fractor::Workflow::DeadLetterQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/workflow/dead_letter_queue.rb

Overview

Dead Letter Queue for capturing permanently failed work

The dead letter queue captures work items that have exhausted all retry attempts and cannot be processed successfully. This provides a mechanism for:

  • Preventing data loss for failed items
  • Enabling manual inspection and reprocessing
  • Supporting different persistence strategies
  • Providing visibility into failure patterns

Examples:

Basic usage

dlq = DeadLetterQueue.new(max_size: 1000)
dlq.add(work, error, context)
failed_items = dlq.all

With handler

dlq = DeadLetterQueue.new
dlq.on_add do |entry|
  Logger.error("Dead letter: #{entry.error}")
end

Defined Under Namespace

Classes: Entry

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_size: nil, persistence: :memory, **persistence_options) ⇒ DeadLetterQueue

Initialize a new dead letter queue

Parameters:

  • max_size (Integer) (defaults to: nil) —

    Maximum queue size (nil for unlimited)

  • persistence (Symbol) (defaults to: :memory) —

    Persistence strategy (:memory, :file, :redis, :database)

  • persistence_options (Hash) —

    Options for persistence backend



61
62
63
64
65
66
67
68
69
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 61

def initialize(max_size: nil, persistence: :memory, **persistence_options)
  @max_size = max_size
  @entries = []
  @handlers = []
  @mutex = Mutex.new
  @persistence = persistence
  @persistence_options = persistence_options
  @persister = create_persister(persistence, persistence_options)
end

Instance Attribute Details

#entries ⇒ Object (readonly)

Returns the value of attribute entries.



54
55
56
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 54

def entries
  @entries
end

#max_size ⇒ Object (readonly)

Returns the value of attribute max_size.



54
55
56
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 54

def max_size
  @max_size
end

Instance Method Details

#add(work, error, context: nil, metadata: {}) ⇒ Entry

Add a failed work item to the dead letter queue

Parameters:

  • work (Fractor::Work) —

    The failed work item

  • error (Exception) —

    The error that caused failure

  • context (Hash) (defaults to: nil) —

    Additional context about the failure

  • metadata (Hash) (defaults to: {}) —

    Additional metadata

Returns:

  • (Entry) —

    The created entry



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 78

def add(work, error, context: nil, metadata: {})
  entry = Entry.new(
    work: work,
    error: error,
    context: context,
    metadata: metadata,
  )

  @mutex.synchronize do
    # Enforce max size if set
    if max_size && @entries.size >= max_size
      # Remove oldest entry
      removed = @entries.shift
      @persister&.remove(removed)
    end

    @entries << entry
    @persister&.persist(entry)
  end

  # Notify handlers
  notify_handlers(entry)

  entry
end

#all ⇒ Array<Entry>

Get all entries in the queue

Returns:

  • (Array<Entry>) —

    All entries



126
127
128
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 126

def all
  @mutex.synchronize { @entries.dup }
end

#by_error_class(error_class) ⇒ Array<Entry>

Get entries for a specific error class

Parameters:

  • error_class (Class) —

    Error class to filter by

Returns:

  • (Array<Entry>) —

    Matching entries



142
143
144
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 142

def by_error_class(error_class)
  filter { |entry| entry.error.is_a?(error_class) }
end

#by_time_range(start_time, end_time = Time.now) ⇒ Array<Entry>

Get entries within a time range

Parameters:

  • start_time (Time) —

    Start of time range

  • end_time (Time) (defaults to: Time.now) —

    End of time range (defaults to now)

Returns:

  • (Array<Entry>) —

    Entries in time range



151
152
153
154
155
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 151

def by_time_range(start_time, end_time = Time.now)
  filter do |entry|
    entry.timestamp >= start_time && entry.timestamp <= end_time
  end
end

#clear ⇒ Integer

Clear all entries from the queue

Returns:

  • (Integer) —

    Number of entries cleared



173
174
175
176
177
178
179
180
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 173

def clear
  @mutex.synchronize do
    count = @entries.size
    @entries.clear
    @persister&.clear
    count
  end
end

#empty? ⇒ Boolean

Check if queue is empty

Returns:

  • (Boolean) —

    True if empty



192
193
194
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 192

def empty?
  size.zero?
end

#enqueue(work, error, context: nil, metadata: {}) ⇒ Entry

Enqueue a failed work item to the dead letter queue (alias for add) Standardized API method name for consistency across queue implementations

Parameters:

  • work (Fractor::Work) —

    The failed work item

  • error (Exception) —

    The error that caused failure

  • context (Hash) (defaults to: nil) —

    Additional context about the failure

  • metadata (Hash) (defaults to: {}) —

    Additional metadata

Returns:

  • (Entry) —

    The created entry



112
113
114
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 112

def enqueue(work, error, context: nil, metadata: {})
  add(work, error, context: context, metadata: metadata)
end

#filter {|Entry| ... } ⇒ Array<Entry>

Get entries matching a filter

Yields:

  • (Entry) —

    Block to filter entries

Returns:

  • (Array<Entry>) —

    Filtered entries



134
135
136
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 134

def filter(&block)
  @mutex.synchronize { @entries.select(&block) }
end

#full? ⇒ Boolean

Check if queue is at capacity

Returns:

  • (Boolean) —

    True if at max_size



199
200
201
202
203
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 199

def full?
  return false unless max_size

  size >= max_size
end

#on_add {|Entry| ... } ⇒ Object

Register a handler to be called when items are added

Yields:

  • (Entry) —

    Block to execute when item is added



119
120
121
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 119

def on_add(&block)
  @handlers << block if block
end

#remove(entry) ⇒ Entry?

Remove an entry from the queue

Parameters:

  • entry (Entry) —

    Entry to remove

Returns:

  • (Entry, nil) —

    Removed entry or nil



161
162
163
164
165
166
167
168
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 161

def remove(entry)
  @mutex.synchronize do
    if @entries.delete(entry)
      @persister&.remove(entry)
      entry
    end
  end
end

#retry_all {|Work| ... } ⇒ Hash

Retry all entries

Yields:

  • (Work) —

    Block to process each work item

Returns:

  • (Hash) —

    Results with :success and :failed counts



247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 247

def retry_all(&block)
  return { success: 0, failed: 0 } unless block

  results = { success: 0, failed: 0 }
  entries_to_retry = all

  entries_to_retry.each do |entry|
    if retry_entry(entry, &block)
      results[:success] += 1
    else
      results[:failed] += 1
    end
  end

  results
end

#retry_entry(entry) {|Work| ... } ⇒ Boolean

Retry a specific entry

Parameters:

  • entry (Entry) —

    Entry to retry

Yields:

  • (Work) —

    Block to process the work

Returns:

  • (Boolean) —

    True if retry succeeded



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 227

def retry_entry(entry, &block)
  return false unless block

  begin
    yield(entry.work)
    remove(entry)
    true
  rescue StandardError => e
    # Add back to queue with new error
    remove(entry)
    add(entry.work, e, context: entry.context,
                       metadata: entry.metadata.merge(retried: true))
    false
  end
end

#size ⇒ Integer

Get the current size of the queue

Returns:

  • (Integer) —

    Number of entries



185
186
187
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 185

def size
  @mutex.synchronize { @entries.size }
end

#stats ⇒ Hash

Get statistics about the queue

Returns:

  • (Hash) —

    Queue statistics



208
209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/fractor/workflow/dead_letter_queue.rb', line 208

def stats
  entries_copy = all

  {
    size: entries_copy.size,
    max_size: max_size,
    full: full?,
    oldest_timestamp: entries_copy.first&.timestamp,
    newest_timestamp: entries_copy.last&.timestamp,
    error_classes: entries_copy.map { |e| e.error.class.name }.uniq,
    persistence: @persistence,
  }
end