RedisBroker: Track non-idempotent jobs as running, too

Push running-job markers down into the Redis for all jobs, and move
the logic for re-enqueueing jobs from dead brokers into the script
itself.  Non-idempotent jobs running on a dead broker are still NOT
re-enqueued.

This should cause non-idempotent jobs to no longer run "invisibly"
on a Redis broker, as well as causing dead brokers to signal any
non-idempotent jobs that were running on them as failed.

Fixes: Issue #18

spinach/brokers/redis_scripts/enqueue_jobs_from_dead_broker.lua

@@ -16,7 +16,7 @@ local jobs_json = redis.call('hvals', running_jobs_key)
 
 for _, job_json in ipairs(jobs_json) do
     local job = cjson.decode(job_json)
-    if job["retries"] < job["max_retries"] then
+    if job["max_retries"] > 0 and job["retries"] < job["max_retries"] then
         job["retries"] = job["retries"] + 1
         -- Set job status to queued:
         -- A major difference between retrying a job failing in a worker and

spinach/brokers/redis_scripts/get_jobs_from_queue.lua

@@ -30,13 +30,11 @@ repeat
         job["status"] = job_status_running
         local job_json = cjson.encode(job)
 
-        if job["max_retries"] > 0 then
-            -- job is idempotent, must track if it's running
-            redis.call('hset', running_jobs_key, job["id"], job_json)
-            -- If tracking concurrency, bump the current value.
-            if max_concurrency ~= -1 then
-                redis.call('hincrby', current_concurrency_key, job['task_name'], 1)
-            end
+        -- track the running job
+        redis.call('hset', running_jobs_key, job["id"], job_json)
+        -- If tracking concurrency, bump the current value.
+        if max_concurrency ~= -1 then
+            redis.call('hincrby', current_concurrency_key, job['task_name'], 1)
         end
 
         jobs[i] = job_json

tests/test_redis_brokers.py

@@ -79,7 +79,11 @@ def test_running_job(broker):
     broker.enqueue_jobs([job])
     assert broker._r.hget(running_jobs_key, str(job.id)) is None
     broker.get_jobs_from_queue('foo_queue', 1)
-    assert broker._r.hget(running_jobs_key, str(job.id)) is None
+    job.status = JobStatus.RUNNING
+    assert (
+        Job.deserialize(broker._r.hget(running_jobs_key, str(job.id)).decode())
+        == job
+    )
     # Try to remove it, even if it doesn't exist in running
     broker.remove_job_from_running(job)
 
@@ -281,8 +285,13 @@ def test_enqueue_jobs_from_dead_broker(broker, broker_2):
     )
     assert current == b'1'
 
-    # Mark broker as dead, should re-enqueue only the idempotent jobs.
-    assert broker_2.enqueue_jobs_from_dead_broker(broker._id) == (2, [])
+    # Mark broker as dead, should re-enqueue only the idempotent jobs.  The
+    # non-idempotent job will report as failed.
+    num_requeued, failed = broker_2.enqueue_jobs_from_dead_broker(broker._id)
+    assert num_requeued == 2
+    jobs = [Job.deserialize(job.decode()) for job in failed]
+    job_1.status = JobStatus.RUNNING
+    assert [job_1] == jobs
 
     # Check that the current_concurrency was decremented for job_3.
     current = broker._r.hget(