Fix connection leak in utils.pipeline() on exception - #4345

What

@contextmanager
def pipeline(redis_obj):
    p = redis_obj.pipeline()
    yield p
    p.execute()

If the code inside with pipeline(redis_obj) as p: raises, p.execute() never runs. That matters because Pipeline.execute() is where the connection gets unconditionally released back to the pool, via its own finally: self.reset().

For a plain pipeline (no watch()), this is low-impact - commands are just buffered into an in-memory list (pipeline_execute_command) until execute() runs, so nothing's actually checked out from the pool yet if the body raises before reaching it. But watch() goes through immediate_execute_command, which checks out a real connection right away (WATCH has to reach the server immediately to establish the watch condition). If the with block then raises before execute() runs - a WatchError on retry, a bug in the caller's own transaction logic, anything - that connection stays checked out until Pipeline.__del__ happens to run during garbage collection, which isn't deterministic or immediate.

Fix

@contextmanager
def pipeline(redis_obj):
    p = redis_obj.pipeline()
    try:
        yield p
        p.execute()
    finally:
        p.reset()

Same call Pipeline.__exit__ already makes unconditionally. Safe to call twice - reset() right after a successful execute() is a no-op, since execute() already reset everything in its own finally.

Testing

Added TestPipelineContextManager in tests/test_utils.py with a fake pipeline tracking execute()/reset() calls - asserts reset() runs on both the success and exception paths, and execute() only runs on success. Verified both new tests fail against the pre-fix code and pass against the fix.

Full tests/test_utils.py: 64 passed. ruff check/ruff format --check clean on both changed files.