Class: Wurk::JobRecord

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/job_record.rb

Overview

One job payload viewed from the data API (Queue#each / JobSet#each). Wraps the raw JSON string from Redis; parses lazily so an O(n) scan over a large queue doesn't pay JSON cost for jobs that go unused.

The value (raw JSON string) is what Redis stores; LREM matches exact bytes, so delete must use it rather than re-serialize.

Spec: docs/target/sidekiq-free.md §19.3.

Direct Known Subclasses

SortedEntry

Constant Summary collapse

ACTIVE_JOB_WRAPPER =

Pre-compiled. ActiveJob's wrapper class varies per Rails minor (ActiveJob::QueueAdapters::SidekiqAdapter::JobWrapper, ActiveJob::QueueAdapters::WurkAdapter::JobWrapper, etc.).

/\AActiveJob::QueueAdapters::.+::JobWrapper\z/
ACTION_MAILER_JOBS = 
%w[
  ActionMailer::DeliveryJob
  ActionMailer::Parameterized::DeliveryJob
  ActionMailer::MailDeliveryJob
].freeze
AJ_PREFIX = 
'_aj_'
AJ_GLOBALID = 
'_aj_globalid'

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(item, queue_name = nil) ⇒ JobRecord

Returns a new instance of JobRecord.

Parameters:

  • item (String, Hash) —

    raw JSON payload or pre-parsed hash.

  • queue_name (String, nil) (defaults to: nil) —

    queue this record came from.



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/wurk/job_record.rb', line 30

def initialize(item, queue_name = nil)
  if item.is_a?(String)
    @value = item
    @item = nil
  else
    @item = item
    @value = nil
  end
  @queue = queue_name
end

Class Method Details

.latency_from(enqueued_at, now_ms = nil) ⇒ Object

Shared latency math: ms ints (>= 10^10) and float secs (< 10^10) both stored in enqueued_at historically. See spec §31.5.



145
146
147
148
149
150
151
152
 
# File 'lib/wurk/job_record.rb', line 145

def self.latency_from(enqueued_at, now_ms = nil)
  return 0.0 if enqueued_at.nil?

  now_ms ||= ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond)
  enq_ms = enqueued_at < 10_000_000_000 ? enqueued_at * 1_000 : enqueued_at
  diff = (now_ms - enq_ms) / 1_000.0
  diff.negative? ? 0.0 : diff
end

Instance Method Details

#[](name) ⇒ Object

Hash-like reader for arbitrary payload fields. Spec §19.3.



84
 
# File 'lib/wurk/job_record.rb', line 84

def [](name) = item[name]

#args ⇒ Object 



62
63
64
65
 
# File 'lib/wurk/job_record.rb', line 62

def args
  parsed = item
  @invalid_args || parsed['args']
end

#bid ⇒ Object 



60
 
# File 'lib/wurk/job_record.rb', line 60

def bid           = item['bid']

#created_at ⇒ Object

Falls back to enqueued_at as upstream does; nil (not upstream's epoch 0) only when both are absent, so a dashboard shows "unknown", not 1970.



79
 
# File 'lib/wurk/job_record.rb', line 79

def created_at    = parse_time(item['created_at'] || item['enqueued_at'])

#delete ⇒ Object

Removes exactly this payload's bytes from the queue list. Returns true when LREM removed ≥1 entry. Idempotent. Method name is Sidekiq wire-compat — renaming would break JobRecord#delete.



109
110
111
112
 
# File 'lib/wurk/job_record.rb', line 109

def delete
  removed = Wurk.redis { |c| c.call('LREM', Keys.queue(@queue), 1, value) }
  removed.to_i.positive?
end

#display_args ⇒ Object

UI-facing args. Encrypted jobs (§4.7) get their envelope last arg masked as "[encrypted data]" so ciphertext never reaches the dashboard; redaction keys off the envelope shape, so it fires whether or not the stored hash carried the encrypt flag. Cleartext preceding args stay visible for triage. Display-only — the stored payload is untouched.

A host redact_args hook (Wurk::Redact) replaces all of that: it sees the whole job and decides what is shown. This is the single choke point for every surface that renders a job — dashboard JSON, search, the /v1 machine API, Sidekiq::Web-style views.



132
133
134
135
136
137
138
139
140
 
# File 'lib/wurk/job_record.rb', line 132

def display_args
  return @display_args if defined?(@display_args)

  hook = Wurk::Redact.hook
  return @display_args = Wurk::Redact.args(item, hook) if hook

  base = active_job_wrapper? ? unwrap_args : args
  @display_args = Wurk::Encryption.redact_args('args' => base, 'encrypt' => item['encrypt'])
end

#display_class ⇒ Object

ActiveJob/ActionMailer unwrappers — UI-facing only. For plain Wurk workers, returns the raw class name.



116
117
118
119
120
 
# File 'lib/wurk/job_record.rb', line 116

def display_class
  return @display_class if defined?(@display_class)

  @display_class = item['display_class'] || (active_job_wrapper? ? unwrap_class : klass)
end

#enqueued_at ⇒ Object 



76
77
78
 
# File 'lib/wurk/job_record.rb', line 76

def enqueued_at   = parse_time(item['enqueued_at'])
# Falls back to enqueued_at as upstream does; nil (not upstream's epoch 0)
# only when both are absent, so a dashboard shows "unknown", not 1970.

#error_backtrace ⇒ Object

Sidekiq compresses the bt as base64(zlib.deflate(JSON.dump(bt))). Returns nil when no error has been recorded.



88
89
90
91
92
93
94
95
96
97
 
# File 'lib/wurk/job_record.rb', line 88

def error_backtrace
  compressed = item['error_backtrace']
  return nil unless compressed

  # Before anything that can raise: the rescue below names Zlib.
  require 'zlib'
  Wurk.load_json(Zlib.inflate(Base64.decode64(compressed)))
rescue Zlib::DataError, ArgumentError, ::JSON::ParserError
  nil
end

#failed_at ⇒ Object 



80
 
# File 'lib/wurk/job_record.rb', line 80

def failed_at     = parse_time(item['failed_at'])

#item ⇒ Object

Lazily parsed payload. Memoized; never re-parses. Anything that is not a JSON object reads as an empty hash whose raw bytes become args, as upstream does for invalid JSON, so one corrupt payload can't break a dashboard page or a Queue#each walk. Valid JSON counts: [1,2] and 123 would otherwise raise from every reader that indexes it by field name.



48
49
50
 
# File 'lib/wurk/job_record.rb', line 48

def item
  @item ||= parse_payload
end

#iterable_state ⇒ Object

IterableJob progress for this job, or nil for a non-iterable job (no it-<jid> HASH). Spec §19.3. Reads via the IterableJobQuery data API.



69
70
71
72
73
 
# File 'lib/wurk/job_record.rb', line 69

def iterable_state
  return nil if jid.nil? || jid.to_s.empty?

  Wurk::IterableJobQuery.new([jid])[jid]
end

#jid ⇒ Object 



59
 
# File 'lib/wurk/job_record.rb', line 59

def jid           = item['jid']

#klass ⇒ Object 



58
 
# File 'lib/wurk/job_record.rb', line 58

def klass         = item['class']

#latency ⇒ Object

Seconds since enqueued_at. Handles legacy float-seconds and current integer-ms enqueued_at shapes. Returns 0.0 when missing or somehow in the future (clock skew).



102
103
104
 
# File 'lib/wurk/job_record.rb', line 102

def latency
  JobRecord.latency_from(item['enqueued_at'] || item['created_at'])
end

#queue ⇒ Object 



41
 
# File 'lib/wurk/job_record.rb', line 41

def queue = @queue || item['queue']

#retried_at ⇒ Object 



81
 
# File 'lib/wurk/job_record.rb', line 81

def retried_at    = parse_time(item['retried_at'])

#tags ⇒ Object 



75
 
# File 'lib/wurk/job_record.rb', line 75

def tags          = item['tags'] || []

#value ⇒ Object

Lazily serialized payload. When constructed from a Hash, we generate JSON on first call so delete (LREM) has exact bytes.



54
55
56
 
# File 'lib/wurk/job_record.rb', line 54

def value
  @value ||= Wurk.dump_json(@item)
end