Module: Wurk::Status

Defined in:
lib/wurk/status.rb,
lib/wurk/status/record.rb,
lib/wurk/status/progress.rb

Overview

Per-job status, progress and result — a Wurk extra, not a Sidekiq surface. Sidekiq loses a job the moment it succeeds: there is no lookup by jid, no progress outside batches and iterables, and no record of what perform returned. Wurk::Status is that record.

Redis schema — one new key, nothing existing touched:

status:<jid>   HASH   state, queue, class, enqueued_at, started_at,
                    finished_at, progress, total, message, result,
                    error_class, error_message, attempt

Every write re-stamps the row's TTL (status_ttl, default Keys::STATUS_TTL), so an abandoned row disappears on its own — no sweeper, no unbounded key growth. A succeeded job's row can be kept longer, or dropped immediately, via status_retention.

Tracking is opt-in per worker class. A class that doesn't opt in never reaches this module, so an untracked job costs exactly what it costs today.

See Also:

Defined Under Namespace

Classes: Progress, Record

Constant Summary collapse

STATES =

The state machine, in order. enqueued and running are transient; the rest are terminal for one attempt (retrying is terminal for the attempt, not the job — the next attempt writes running again).

%w[enqueued running complete failed interrupted retrying dead].freeze

Class Method Summary collapse

Class Method Details

.default_ttl(config = Wurk.configuration) ⇒ Integer

Returns seconds a status row lives, re-stamped on every write.

Returns:

  • (Integer) —

    seconds a status row lives, re-stamped on every write.



46
 
# File 'lib/wurk/status.rb', line 46

def default_ttl(config = Wurk.configuration) = (config[:status_ttl] || Keys::STATUS_TTL).to_i

.delete(jid, pool: nil) ⇒ Boolean

Returns true when a row was actually removed.

Returns:

  • (Boolean) —

    true when a row was actually removed.



97
98
99
 
# File 'lib/wurk/status.rb', line 97

def delete(jid, pool: nil) # rubocop:disable Naming/PredicateMethod
  with_pool(pool) { |conn| conn.call('UNLINK', key(jid)) }.to_i.positive?
end

.enqueued(pipe, job, at_millis, ttl: default_ttl) ⇒ Object

Append this job's enqueued row onto an already-open pipeline — the client's own queue-write pipeline, never one of ours. Two commands, no round trip of its own, and nothing at all for a job whose class never opted in: Client does not call this for an untracked payload, so a plain push stays exactly the SADD + LPUSH it has always been.

HSET + EXPIRE rather than the status_write script, for two reasons. A NOSCRIPT from an EVALSHA surfaces only when the pipeline finalizes, and Client#push_immediate keeps Lua out of the plain pipeline precisely so a script reload can never replay an already-applied LPUSH into a second copy of the job; both commands here are idempotent, so they are also safe in the batched pipeline, which does replay. And the create gate the script exists for is meaningless on this path: this write IS the create.

Only immediate pushes write a row. A scheduled job sits in the ZSET for minutes or days — well past #default_ttl — so a row claiming enqueued would expire before the job ran, and the state machine has no scheduled state to tell the truth with. Its first row is the running one the server middleware writes, which re-derives every timestamp from the payload anyway.



76
77
78
79
80
81
82
83
84
 
# File 'lib/wurk/status.rb', line 76

def enqueued(pipe, job, at_millis, ttl: default_ttl)
  jid = job['jid']
  return if jid.nil? || jid.to_s.empty?

  row = key(jid)
  pipe.call('HSET', row, 'state', 'enqueued', 'class', job['class'].to_s,
            'queue', job['queue'].to_s, 'enqueued_at', (at_millis / 1000.0).to_s)
  pipe.call('EXPIRE', row, ttl)
end

.get(jid, pool: nil) ⇒ Wurk::Status::Record?

Returns nil when no row exists — the jid is unknown, the class isn't tracked, or the row's TTL has lapsed.

Returns:

  • (Wurk::Status::Record, nil) —

    nil when no row exists — the jid is unknown, the class isn't tracked, or the row's TTL has lapsed.



88
89
90
91
92
93
94
 
# File 'lib/wurk/status.rb', line 88

def get(jid, pool: nil)
  raw = with_pool(pool, idempotent: true) { |conn| conn.call('HGETALL', key(jid)) }
  row = normalize_hgetall(raw)
  return nil if row.empty?

  Record.new(jid.to_s, row)
end

.key(jid) ⇒ Object 



37
 
# File 'lib/wurk/status.rb', line 37

def key(jid) = Keys.status(jid)

.retention(config = Wurk.configuration) ⇒ Integer?

How long a complete row outlives the job that wrote it.

Returns:

  • (Integer, nil) —

    nil (the default) to expire on the same clock as every other row; 0 to delete the row the moment the job succeeds, which is what Sidekiq does today — a succeeded job leaves nothing.



53
 
# File 'lib/wurk/status.rb', line 53

def retention(config = Wurk.configuration) = config[:status_retention]&.to_i

.tracked?(job) ⇒ Boolean

The opt-in gate, in one place: everything that writes a status row asks this first. track rides on the job payload (merged in from the class's sidekiq_options), so a job enqueued before the option was set keeps running untracked to completion instead of half-tracking.

Returns:

  • (Boolean) 


43
 
# File 'lib/wurk/status.rb', line 43

def tracked?(job) = job['track'] ? true : false

.write(jid, ttl: nil, create: true, pool: nil, **fields) ⇒ Boolean

Write fields onto a status row and re-stamp its TTL, in one round trip.

create: false is the progress path: it updates a live row but will not conjure one that expired or was deleted, which would leave a stateless phantom behind (see lua/status_write.lua).

nil values are dropped rather than written as empty strings, so a caller can pass the whole field set every time and let the unset ones fall away.

Returns:

  • (Boolean) —

    true when the row was written.



112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/wurk/status.rb', line 112

def write(jid, ttl: nil, create: true, pool: nil, **fields) # rubocop:disable Naming/PredicateMethod
  validate_state!(fields[:state])
  argv = flatten(fields)
  return false if argv.empty?

  wrote = with_pool(pool) do |conn|
    Lua::Loader.eval_cached(conn, :status_write, keys: [key(jid)],
                                                 argv: [(ttl || default_ttl).to_i, create ? '1' : '0', *argv])
  end
  wrote.to_i == 1
end