     nil                   - Already GC'd or invalid message id."
   [qname mid]
   (car/parse-keyword
-   (car/lua ; Careful, logic here's quite subtle: see state diagram for assistance
-    (format "%s\n%s" lua-message-status "return state")
+   (car/lua
+    (script-with-msg-status :mid-arg nil "return state")
     {:qk-messages (qkey qname :messages)
      :qk-locks    (qkey qname :locks)
      :qk-backoffs (qkey qname :backoffs)

              (integer? eoq-backoff-ms) (mapv (constantly eoq-backoff-ms) (range 5))
              :else (throw (Exception. (str "Bad eoq-backoff-ms: " eoq-backoff-ms))))]
    (car/lua
     "if redis.call('exists', _:qk-eoq-backoff) == 1 then return 'eoq-backoff' end

     -- TODO Waiting for Lua brpoplpush support to get us long polling
     local mid = redis.call('rpoplpush', _:qk-mid-circle, _:qk-mid-circle)
     local now = tonumber(_:now)

     if (not mid) or (mid == 'end-of-circle') then -- Uninit'd or eoq

       -- Calculate eoq_backoff_ms
       local ndry_runs = tonumber(redis.call('get', _:qk-ndry-runs) or 0)
       local eoq_ms_tab = {_:eoq-ms0, _:eoq-ms1, _:eoq-ms2, _:eoq-ms3, _:eoq-ms4}
       local eoq_backoff_ms = tonumber(eoq_ms_tab[math.min(5, (ndry_runs + 1))])

       -- Set queue-wide polling backoff flag
       redis.call('psetex', _:qk-eoq-backoff, eoq_backoff_ms, 'true')
       redis.call('incr',   _:qk-ndry-runs)

       return 'eoq-backoff'
     end

     local lock_exp    = tonumber(redis.call('hget', _:qk-locks,    mid) or 0)
     local backoff_exp = tonumber(redis.call('hget', _:qk-backoffs, mid) or 0)

     -- Active lock/backoff
     if (now < lock_exp) or (now < backoff_exp) then return nil end

     redis.call('set', _:qk-ndry-runs, 0) -- Doing useful work

     if redis.call('sismember', _:qk-done, mid) == 1 then -- GC
       redis.call('hdel',  _:qk-messages,   mid)
       redis.call('hdel',  _:qk-locks,      mid)
       redis.call('hdel',  _:qk-backoffs,   mid)
       redis.call('hdel',  _:qk-nattempts,  mid)
       redis.call('ltrim', _:qk-mid-circle, 1, -1)
       redis.call('srem',  _:qk-done,       mid)
       return nil
     end

     redis.call('hset', _:qk-locks, mid, now + tonumber(_:lock-ms)) -- Acquire
     local mcontent  = redis.call('hget',    _:qk-messages,  mid)
     local nattempts = redis.call('hincrby', _:qk-nattempts, mid, 1)
     return {mid, mcontent, nattempts}"
     {:qk-messages    (qkey qname :messages)
      :qk-locks       (qkey qname :locks)
      :qk-backoffs    (qkey qname :backoffs)

  [qname message & [unique-message-id allow-requeue?]]
  (->>
   (car/lua
    (script-with-msg-status :mid-arg nil
     "--
     if (_:allow-requeue? == 'true') and
        ((state == 'locked') or (state == 'done-with-backoff')) and
