<commit_before>-------------------------------------------------------------------------------
-- Job Class
--
-- It returns an object that represents the job with the provided JID
-------------------------------------------------------------------------------

-- This gets all the data associated with the job with the provided id. If the
-- job is not found, it returns nil. If found, it returns an object with the
-- appropriate properties
function QlessJob:data(...)
  local job = redis.call(
      'hmget', QlessJob.ns .. self.jid, 'jid', 'klass', 'state', 'queue',
      'worker', 'priority', 'expires', 'retries', 'remaining', 'data',
      'tags', 'failure', 'spawned_from_jid')

  -- Return nil if we haven't found it
  if not job[1] then
    return nil
  end

  local data = {
    jid              = job[1],
    klass            = job[2],
    state            = job[3],
    queue            = job[4],
    worker           = job[5] or '',
    tracked          = redis.call(
      'zscore', 'ql:tracked', self.jid) ~= false,
    priority         = tonumber(job[6]),
    expires          = tonumber(job[7]) or 0,
    retries          = tonumber(job[8]),
    remaining        = math.floor(tonumber(job[9])),
    data             = job[10],
    tags             = cjson.decode(job[11]),
    history          = self:history(),
    failure          = cjson.decode(job[12] or '{}'),
    spawned_from_jid = job[13],
    dependents       = redis.call(
      'smembers', QlessJob.ns .. self.jid .. '-dependents'),
    dependencies     = redis.call(
      'smembers', QlessJob.ns .. self.jid .. '-dependencies')
  }

  if #arg > 0 then
    -- This section could probably be optimized, but I wanted the interface
    -- in place first
    local response = {}
    for index, key in ipairs(arg) do
      table.insert(response, data[key])
    end
    return response
  else
    return data
  end
end

-- Complete a job and optionally put it in another queue, either scheduled or
-- to be considered waiting immediately. It can also optionally accept other
-- jids on which this job will be considered dependent before it's considered 
-- valid.
--
-- The variable-length arguments may be pairs of the form:
-- 
--      ('next'   , queue) : The queue to advance it to next
--      ('delay'  , delay) : The delay for the next queue
--      ('depends',        : Json of jobs it depends on in the new queue
--          '["jid1", "jid2", ...]')
---
function QlessJob:complete(now, worker, queue, data, ...)
  assert(worker, 'Complete(): Arg "worker" missing')
  assert(queue , 'Complete(): Arg "queue" missing')
  data = assert(cjson.decode(data),
    'Complete(): Arg "data" missing or not JSON: ' .. tostring(data))

  -- Read in all the optional parameters
  local options = {}
  for i = 1, #arg, 2 do options[arg[i]] = arg[i + 1] end
  
  -- Sanity check on optional args
  local nextq   = options['next']
  local delay   = assert(tonumber(options['delay'] or 0))
  local depends = assert(cjson.decode(options['depends'] or '[]'),
    'Complete(): Arg "depends" not JSON: ' .. tostring(options['depends']))

  -- Depends doesn't make sense without nextq
  if options['delay'] and nextq == nil then
    error('Complete(): "delay" cannot be used without a "next".')
  end

  -- Depends doesn't make sense without nextq
  if options['depends'] and nextq == nil then
    error('Complete(): "depends" cannot be used without a "next".')
  end

  -- The bin is midnight of the provided day
  -- 24 * 60 * 60 = 86400
  local bin = now - (now % 86400)

  -- First things first, we should see if the worker still owns this job
  local lastworker, state, priority, retries, current_queue = unpack(
    redis.call('hmget', QlessJob.ns .. self.jid, 'worker', 'state',
      'priority', 'retries', 'queue'))

  if lastworker == false then
    error('Complete(): Job does not exist')
  elseif (state ~= 'running') then
    error('Complete(): Job is not currently running: ' .. state)
  elseif lastworker ~= worker then
    error('Complete(): Job has been handed out to another worker: ' ..
      tostring(lastworker))
  elseif queue ~= current_queue then
    error('Complete(): Job running in another queue: ' ..
      tostring(current_queue))
  end

  -- Now we can assume that the worker does own the job. We need to
  --    1) Remove the job from the 'locks' from the old queue
  --    2) Enqueue it in the next stage if necessary
  --    3) Update the data
  --    4) Mark the job as completed, remove the worker, remove expires, and
  --          update history
  self:history(now, 'done')

  if data then
    redis.call('hset', QlessJob.ns .. self.jid, 'data', cjson.encode(data))
  end

  -- Remove the job from the previous queue
  local queue_obj = Qless.queue(queue)
  queue_obj.work.remove(self.jid)
  queue_obj.locks.remove(self.jid)
  queue_obj.scheduled.remove(self.jid)

  ----------------------------------------------------------
  -- This is the massive stats update that we have to do
  ----------------------------------------------------------
  -- This is how long we've been waiting to get popped
  -- local waiting = math.floor(now) - history[#history]['popped']
  local time = tonumber(
    redis.call('hget', QlessJob.ns .. self.jid, 'time') or now)
  local waiting = now - time
  Qless.queue(queue):stat(now, 'run', waiting)
  redis.call('hset', QlessJob.ns .. self.jid,
    'time', string.format("%.20f", now))

  -- Remove this job from the jobs that the worker that was running it has
  redis.call('zrem', 'ql:w:' .. worker .. ':jobs', self.jid)

  if redis.call('zscore', 'ql:tracked', self.jid) ~= false then
    Qless.publish('completed', self.jid)
  end

  if nextq then
    queue_obj = Qless.queue(nextq)
    -- Send a message out to log
    Qless.publish('log', cjson.encode({<commit_msg>Fix bug where empty arrays in job data were being converted into objects when advancing<commit_after>-------------------------------------------------------------------------------
-- Job Class
--
-- It returns an object that represents the job with the provided JID
-------------------------------------------------------------------------------

-- This gets all the data associated with the job with the provided id. If the
-- job is not found, it returns nil. If found, it returns an object with the
-- appropriate properties
function QlessJob:data(...)
  local job = redis.call(
      'hmget', QlessJob.ns .. self.jid, 'jid', 'klass', 'state', 'queue',
      'worker', 'priority', 'expires', 'retries', 'remaining', 'data',
      'tags', 'failure', 'spawned_from_jid')

  -- Return nil if we haven't found it
  if not job[1] then
    return nil
  end

  local data = {
    jid              = job[1],
    klass            = job[2],
    state            = job[3],
    queue            = job[4],
    worker           = job[5] or '',
    tracked          = redis.call(
      'zscore', 'ql:tracked', self.jid) ~= false,
    priority         = tonumber(job[6]),
    expires          = tonumber(job[7]) or 0,
    retries          = tonumber(job[8]),
    remaining        = math.floor(tonumber(job[9])),
    data             = job[10],
    tags             = cjson.decode(job[11]),
    history          = self:history(),
    failure          = cjson.decode(job[12] or '{}'),
    spawned_from_jid = job[13],
    dependents       = redis.call(
      'smembers', QlessJob.ns .. self.jid .. '-dependents'),
    dependencies     = redis.call(
      'smembers', QlessJob.ns .. self.jid .. '-dependencies')
  }

  if #arg > 0 then
    -- This section could probably be optimized, but I wanted the interface
    -- in place first
    local response = {}
    for index, key in ipairs(arg) do
      table.insert(response, data[key])
    end
    return response
  else
    return data
  end
end

-- Complete a job and optionally put it in another queue, either scheduled or
-- to be considered waiting immediately. It can also optionally accept other
-- jids on which this job will be considered dependent before it's considered 
-- valid.
--
-- The variable-length arguments may be pairs of the form:
-- 
--      ('next'   , queue) : The queue to advance it to next
--      ('delay'  , delay) : The delay for the next queue
--      ('depends',        : Json of jobs it depends on in the new queue
--          '["jid1", "jid2", ...]')
---
function QlessJob:complete(now, worker, queue, raw_data, ...)
  assert(worker, 'Complete(): Arg "worker" missing')
  assert(queue , 'Complete(): Arg "queue" missing')
  local data = assert(cjson.decode(raw_data),
    'Complete(): Arg "data" missing or not JSON: ' .. tostring(raw_data))

  -- Read in all the optional parameters
  local options = {}
  for i = 1, #arg, 2 do options[arg[i]] = arg[i + 1] end
  
  -- Sanity check on optional args
  local nextq   = options['next']
  local delay   = assert(tonumber(options['delay'] or 0))
  local depends = assert(cjson.decode(options['depends'] or '[]'),
    'Complete(): Arg "depends" not JSON: ' .. tostring(options['depends']))

  -- Depends doesn't make sense without nextq
  if options['delay'] and nextq == nil then
    error('Complete(): "delay" cannot be used without a "next".')
  end

  -- Depends doesn't make sense without nextq
  if options['depends'] and nextq == nil then
    error('Complete(): "depends" cannot be used without a "next".')
  end

  -- The bin is midnight of the provided day
  -- 24 * 60 * 60 = 86400
  local bin = now - (now % 86400)

  -- First things first, we should see if the worker still owns this job
  local lastworker, state, priority, retries, current_queue = unpack(
    redis.call('hmget', QlessJob.ns .. self.jid, 'worker', 'state',
      'priority', 'retries', 'queue'))

  if lastworker == false then
    error('Complete(): Job does not exist')
  elseif (state ~= 'running') then
    error('Complete(): Job is not currently running: ' .. state)
  elseif lastworker ~= worker then
    error('Complete(): Job has been handed out to another worker: ' ..
      tostring(lastworker))
  elseif queue ~= current_queue then
    error('Complete(): Job running in another queue: ' ..
      tostring(current_queue))
  end

  -- Now we can assume that the worker does own the job. We need to
  --    1) Remove the job from the 'locks' from the old queue
  --    2) Enqueue it in the next stage if necessary
  --    3) Update the data
  --    4) Mark the job as completed, remove the worker, remove expires, and
  --          update history
  self:history(now, 'done')

  if raw_data then
    redis.call('hset', QlessJob.ns .. self.jid, 'data', raw_data)
  end

  -- Remove the job from the previous queue
  local queue_obj = Qless.queue(queue)
  queue_obj.work.remove(self.jid)
  queue_obj.locks.remove(self.jid)
  queue_obj.scheduled.remove(self.jid)

  ----------------------------------------------------------
  -- This is the massive stats update that we have to do
  ----------------------------------------------------------
  -- This is how long we've been waiting to get popped
  -- local waiting = math.floor(now) - history[#history]['popped']
  local time = tonumber(
    redis.call('hget', QlessJob.ns .. self.jid, 'time') or now)
  local waiting = now - time
  Qless.queue(queue):stat(now, 'run', waiting)
  redis.call('hset', QlessJob.ns .. self.jid,
    'time', string.format("%.20f", now))

  -- Remove this job from the jobs that the worker that was running it has
  redis.call('zrem', 'ql:w:' .. worker .. ':jobs', self.jid)

  if redis.call('zscore', 'ql:tracked', self.jid) ~= false then
    Qless.publish('completed', self.jid)
  end

  if nextq then
    queue_obj = Qless.queue(nextq)
    -- Send a message out to log
    Qless.publish('log', cjson.encode({
{'fraction_non_alphanumeric': 0.17268217584673282, 'fraction_numerical': 0.008210742387957578, 'mean_word_length': 2.267113718915898, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6654552', 'n_tokens_mistral': 3534, 'n_tokens_neox': 3442, 'n_words': 1236}