`finish_failed_job` ignores a function's `keep_result` / `keep_result_forever` (only `finish_job` honours them)

[ericz7900]

Worker.finish_job resolves the result TTL from the function first
(function.keep_result_s / function.keep_result_forever, falling back to the worker defaults),
but Worker.finish_failed_job only ever reads the worker-level self.keep_result_s /
self.keep_result_forever:

# arq/worker.py, v0.28.0
async def finish_failed_job(self, job_id: str, result_data: Optional[bytes]) -> None:
    ...
    keep_result = self.keep_result_forever or self.keep_result_s > 0
    if result_data is not None and keep_result:
        expire = 0 if self.keep_result_forever else self.keep_result_s
        tr.set(result_key_prefix + job_id, result_data, px=to_ms(expire))

So a function registered with func(fn, keep_result=0) still leaves arq:result:<job_id> behind
for the worker default (3600 s) whenever the job fails terminally — including the
job_try > max_tries branch, which returns before after_job_end, so there is no hook to clean
it up either. Because enqueue_job refuses a _job_id while a result key for it exists, an
explicit "no result window" on the function silently fails to suppress dedup-blocking for the
whole default hour — precisely in the retry-exhausted case where re-enqueueing matters most.
The same asymmetry applies to keep_result_forever=True on the function with keep_result=0 on
the worker: the failed result is dropped instead of kept.

Related, in the same function: with keep_result_forever=True on the worker, a job that
exhausts max_tries makes finish_failed_job raise, because expire = 0 if self.keep_result_forever
becomes SET … PX 0, which redis rejects:

redis.exceptions.ResponseError: Command # 4 (SET arq:result:testing … PX 0) of pipeline caused
error: invalid expire time in 'set' command

finish_job uses expire = None for the forever case.

Minimal repro

import asyncio
from arq import Worker, func
from arq.connections import create_pool, RedisSettings
from arq.worker import Retry

async def always_retry(ctx, x):
    raise Retry(defer=0)

async def main():
    pool = await create_pool(RedisSettings())
    w = Worker(functions=[func(always_retry, keep_result=0)], redis_pool=pool,
               max_tries=2, burst=True, poll_delay=0.01)   # worker keep_result left at 3600
    for _ in range(3):
        if await pool.enqueue_job("always_retry", 1, _job_id="dedup") is None:
            break
        await w.main(); await asyncio.sleep(0.02)
    print("re-enqueue accepted:", await pool.enqueue_job("always_retry", 1, _job_id="dedup") is not None)
    print("result key pttl:", await pool.pttl("arq:result:dedup"))

asyncio.run(main())

Prints re-enqueue accepted: False / result key pttl: ~3600000.
Expected, given keep_result=0 on the function: True / -2.

Suggested fix

Pass the Function into finish_failed_job from the max_tries branch and resolve
keep_result_s / keep_result_forever the same way finish_job does, falling back to the worker
defaults only when the function is unknown (job expired / deserialization failed / function not
found); and use expire = None (not 0) for the forever case. Happy to open a PR.