enqueue_job drops the caller's pipeline for rate-limited jobs, so a rolled-back job still runs

What happens

Queue.enqueue_job() takes a pipeline and passes it to _enqueue_job(), but the rate-limited branch one line above drops it:

if job.has_rate_limit:
    return self._enqueue_rate_limited_job(job, at_front=at_front)
return self._enqueue_job(job, pipeline=pipeline, at_front=at_front, unique=unique)

_enqueue_rate_limited_job then sees pipeline is None, opens its own pipeline, executes it, and calls acquire_and_enqueue. So the job is written and pushed onto the queue immediately, inside what the caller believes is an open transaction.

If the caller discards that transaction, the job still runs.

Reproduction

from redis import Redis
from rq import Queue
from rq.rate_limit import RateLimit

conn = Redis(db=15)
conn.flushdb()
q = Queue('rl-repro', connection=conn)

pipe = conn.pipeline()
job = q.enqueue_call(say_hello, rate_limit=RateLimit(key='k', concurrency=5), pipeline=pipe)
pipe.reset()   # caller discards the transaction

print(conn.exists(job.key))  # 1
print(job.get_status(refresh=True))  # JobStatus.QUEUED
print(q.count)  # 1

Same call without rate_limit= writes nothing and leaves q.count at 0, which is the behaviour I expected in both cases.

On 2.12.0 and on master at 0e9ffe2.

Why I think it is an oversight rather than a design choice

_enqueue_rate_limited_job already supports a caller-owned pipeline, and its docstring spells out the contract: buffer the ops, let the caller EXEC, then call acquire_and_enqueue so promotion observes committed state. Both other call sites do exactly that, registry.py:621 and scheduler.py:271. enqueue_job is the only place that has a pipeline in hand and does not pass it on.

The fix, and the part worth your opinion

Passing it through is one line and makes the discarded-transaction case match the unlimited one.

The trade-off is that a caller doing enqueue_call(..., rate_limit=..., pipeline=pipe) followed by pipe.execute() then gets a job sitting in rate_limited until something calls acquire_and_enqueue, rather than one already on the queue. That is the documented contract for the pipeline path and RateLimitRegistry.cleanup() picks such jobs up during maintenance, but it is a visible change for anyone using that combination today, so I would rather you called it than assume.

Happy to send the patch, I have it working with tests.