Module: Wurk::JobUtil

Included in:
Client, Collapse::ClientMiddleware, Flow::Creation, Worker::Setter
Defined in:
lib/wurk/job_util.rb

Overview

Mixin shared by Wurk::Client (and Wurk::Job::Setter) to validate, normalize, and JSON-verify job payloads before they hit Redis.

Spec: docs/target/sidekiq-free.md §9 (Sidekiq::JobUtil).

Constant Summary collapse

TRANSIENT_ATTRIBUTES =

Top-level keys consumed at enqueue time but stripped from every payload before raw_push — they must never reach the wire (spec §2.2):

`pool`         selects the Redis pool (resolved in client_push/build_client)
`client_class` swaps the enqueue client (Wurk.transactional_push!)

Both carry non-JSON values (a pool / a Class). Baked into the literal rather than appended at load: a load-time << is fragile under the parallel test runner (a test that add/deletes the same key clobbers it for later suites). Still mutable so other extensions can append without monkey-patching.

%w[pool client_class]
RETRY_FOR_MAX =

rubocop:disable Style/MutableConstant

1_000_000_000
TRACK_VALUES =

The only values track (the Wurk::Status opt-in) may hold. Checked rather than coerced because Status.tracked? asks the payload for truthiness: track: 'false' would track every job of the class, and track: :maybe would too. Both read as "off" to whoever wrote them and cost a Redis row per job until someone goes looking in the key space.

[true, false, nil].freeze
BOUND_OPTIONS =

The two wall-clock bounds, both in seconds: timeout bounds one attempt, deadline bounds the job from enqueue onward. Checked here rather than where they are read, for the same reason track is — the reader is Middleware::Timeout, on a server one deploy away from whoever wrote deadline: '5 minutes', and its answer to a bound it cannot use is to run the job unbounded, because the payload may predate the option or come from stock Sidekiq. Silence is right there and useless here, so both doors a Wurk-written bound comes through — a class-level sidekiq_options (worker or ActiveJob) and a payload push — call this instead.

%w[timeout deadline].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.positive_seconds?(value) ⇒ Boolean

The one definition of a bound Wurk can act on, shared by the check above and by the middleware that arms it, so "valid where it was declared" and "usable where it is read" cannot drift apart. ActiveSupport durations pass: 5.minutes.is_a?(Numeric) is true by that class's own design.

Returns:

  • (Boolean) 


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

def self.positive_seconds?(value)
  return false unless value.is_a?(::Numeric)

  seconds = value.to_f
  seconds.finite? && seconds.positive?
end

.scheduled_member(hash) ⇒ Object

The exact bytes a job is stored as inside the schedule ZSET. at is dropped because it is the score, and enqueued_at because it marks arrival on an immediate queue — the promoter restamps it on the way out (Lua RELIABLE_SCHEDULE_PROMOTE), so a stored member carrying a stale one would ship the wrong value to any reader that looks before promotion.

One definition, because more than one writer puts members in that ZSET (Client#push_scheduled and Debounce) and wire-compat does not survive the two of them drifting.



85
86
87
 
# File 'lib/wurk/job_util.rb', line 85

def self.scheduled_member(hash)
  Wurk.dump_json(hash.except('enqueued_at', 'at'))
end

.validate_bounds!(item) ⇒ Object

Raises:

  • (ArgumentError) —

    unless every bound present is a positive, finite number of seconds.



56
57
58
59
60
61
62
63
 
# File 'lib/wurk/job_util.rb', line 56

def self.validate_bounds!(item)
  BOUND_OPTIONS.each do |name|
    value = item[name]
    next if value.nil? || positive_seconds?(value)

    raise(ArgumentError, "Job '#{name}' must be a positive number of seconds: `#{item}`")
  end
end

.validate_track!(value, subject) ⇒ Object

Both doors track can come through — a class-level sidekiq_options (worker or ActiveJob) and a raw payload push — end up here, so the two DSL copies stay in sync by calling the same check rather than repeating it. Rejected where it was written, not where it is read.

Raises:

  • (ArgumentError) —

    unless value is true, false, or absent.



37
38
39
40
41
 
# File 'lib/wurk/job_util.rb', line 37

def self.validate_track!(value, subject)
  return if TRACK_VALUES.include?(value)

  raise(ArgumentError, "Job 'track' must be true or false: `#{subject}`")
end

Instance Method Details

#normalize_item(item) ⇒ Object

Validate → merge class/default options → stringify → assign jid & created_at → strip transient keys. Returns the canonical payload.



113
114
115
116
117
118
119
 
# File 'lib/wurk/job_util.rb', line 113

def normalize_item(item)
  validate(item)
  normalized = class_defaults_for(item['class']).merge(item)
  normalized = wrap_options(normalized)
  stringify_identity!(normalized, item['class'])
  finalize(normalized)
end

#now_in_millis ⇒ Object 



121
122
123
 
# File 'lib/wurk/job_util.rb', line 121

def now_in_millis
  ::Process.clock_gettime(::Process::CLOCK_REALTIME, :millisecond)
end

#validate(item) ⇒ Object

Raises:

  • (ArgumentError) —

    if the payload is structurally invalid.



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

def validate(item)
  raise(ArgumentError, "Job must be a Hash with 'class' and 'args' keys: `#{item}`") unless valid_shape?(item)
  raise(ArgumentError, "Job args must be an Array: `#{item}`") unless item['args'].is_a?(Array)
  raise(ArgumentError, "Job class must be a Class or String: `#{item}`") unless valid_class?(item['class'])
  raise(ArgumentError, "Job 'at' must be a Numeric timestamp: `#{item}`") unless valid_at?(item)

  validate_option_values(item)
end

#verify_json(item) ⇒ Object

Walk args; report the first non-JSON-native value according to the configured strict mode. Hash keys must be Strings.



101
102
103
104
105
106
107
108
109
 
# File 'lib/wurk/job_util.rb', line 101

def verify_json(item)
  mode = Wurk.strict_args_mode
  return if mode == false

  offender = json_unsafe(item['args'])
  return if offender.nil?

  report_unsafe(item, offender, mode)
end