Add Redis Streams option for job delivery

[ajac-zero]

This pull request adds a basic implementation of Redis Streams, in order to avoid polling for new jobs in the worker and reduce latency, in accordance with objective 4 of issue #437.

To create a worker that listens to a Redis Stream, we can use the cli or specify it in the code directly.

CLI:
arq worker.WorkerSettings --stream

Code:

class WorkerSettings:
    functions = [...]
    stream = True
    ...

On the client, they must specify that they want to deliver a job to a worker through a Redis Stream.

redis = await create_pool(RedisSettings())
await redis.enqueue_job('hello_world', _use_stream=True)

Here are the results of a very simple benchmark that showcases the potential of using Redis Streams for improved latency.

Polling:

Average time: 0.268s

Streaming:

Average time: 0.012s

[codecov]

Codecov Report

Attention: Patch coverage is 50.00000% with 17 lines in your changes are missing coverage. Please review.

Project coverage is 95.00%. Comparing base (94cd878) to head (1b44875).
Report is 11 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #451      +/-   ##
==========================================
- Coverage   96.27%   95.00%   -1.28%     
==========================================
  Files          11       11              
  Lines        1074     1100      +26     
  Branches      209      197      -12     
==========================================
+ Hits         1034     1045      +11     
- Misses         19       31      +12     
- Partials       21       24       +3     

Files	Coverage Δ
arq/constants.py	100.00% <100.00%> (ø)
arq/cli.py	96.49% <60.00%> (-3.51%)	⬇️
arq/connections.py	88.81% <33.33%> (-1.25%)	⬇️
arq/worker.py	94.88% <43.47%> (-2.29%)	⬇️

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 1315583...1b44875. Read the comment docs.

[gaby]

@ajac-zero This may need a unit-test with stream enabled.

[ajac-zero]

Sure thing @gaby . I was wondering how I should go about that...

What I started doing was add a stream parameter to the worker tests, and then wrap them so they run twice, once with stream and once without.

But I feel this might not be the best way to do things, maybe I should focus on some vital tests? What do you suggest?

[gaby]

@ajac-zero That's probably a great starting point, running current tests with "stream" set to false. Then running the test suite with "stream" set to True. This will require setting the source of the data to use Streams.