feat(worker): worker.run() is now context managed

[shijiesheng]

What changed?

worker.run is now a contextmanager and yields self when it's started and cancels the coroutines on exit

Why?

make it easier to use worker

How did you test it?

Unit Test

Potential risks

Release notes

Documentation Changes

[gitar-bot]

⚠️ Bug: Client context manager dropped in integration test helper

The old code used async with self.client() as client: which properly called Client.__aenter__() (initializing the gRPC channel via ready()) and Client.__aexit__() (closing the channel via close()). The new code passes self.client() directly to Worker(...) without entering the context manager, so:

1. ready() is never called — the client may not be properly initialized.
2. close() is never called — gRPC channel resources will leak after each test.

This will likely cause integration test failures or resource leaks.

Suggested fix:

@asynccontextmanager
async def worker(
    self, registry: Registry, **kwargs: Unpack[WorkerOptions]
) -> AsyncGenerator[Worker, None]:
    async with self.client() as client:
        worker = Worker(client, self.test_name, registry, **kwargs)
        async with worker.run():
            yield worker

_{Was this helpful? React with 👍 / 👎 | Reply gitar fix to apply this suggestion}

[gitar-bot]

💡 Edge Case: Worker tasks may fail silently before yield completes

In Worker.run(), tasks are created via asyncio.create_task() and then yield self is reached. If a task raises an exception before the caller's async with block body executes (e.g., connection failure on first poll), that exception is silently swallowed by return_exceptions=True in the finally block's asyncio.gather(). With the old TaskGroup approach, such exceptions would propagate. Consider checking task results after gather and re-raising if any contain non-CancelledError exceptions.

_{Was this helpful? React with 👍 / 👎 | Reply gitar fix to apply this suggestion}

[gitar-bot]

Code Review ⚠️ Changes requested 0 resolved / 2 findings

Adds context manager support to worker.run() but the integration test helper drops the client context manager, and worker tasks may fail silently before yield completes.

⚠️ Bug: Client context manager dropped in integration test helper

📄 tests/integration_tests/helper.py:25-27

The old code used async with self.client() as client: which properly called Client.__aenter__() (initializing the gRPC channel via ready()) and Client.__aexit__() (closing the channel via close()). The new code passes self.client() directly to Worker(...) without entering the context manager, so:

1. ready() is never called — the client may not be properly initialized.
2. close() is never called — gRPC channel resources will leak after each test.

This will likely cause integration test failures or resource leaks.

Suggested fix

@asynccontextmanager
async def worker(
    self, registry: Registry, **kwargs: Unpack[WorkerOptions]
) -> AsyncGenerator[Worker, None]:
    async with self.client() as client:
        worker = Worker(client, self.test_name, registry, **kwargs)
        async with worker.run():
            yield worker

💡 Edge Case: Worker tasks may fail silently before yield completes

📄 cadence/worker/_worker.py:43-50

In Worker.run(), tasks are created via asyncio.create_task() and then yield self is reached. If a task raises an exception before the caller's async with block body executes (e.g., connection failure on first poll), that exception is silently swallowed by return_exceptions=True in the finally block's asyncio.gather(). With the old TaskGroup approach, such exceptions would propagate. Consider checking task results after gather and re-raising if any contain non-CancelledError exceptions.

🤖 Prompt for agents

Code Review: Adds context manager support to worker.run() but the integration test helper drops the client context manager, and worker tasks may fail silently before yield completes.

1. ⚠️ Bug: Client context manager dropped in integration test helper
   Files: tests/integration_tests/helper.py:25-27

   The old code used `async with self.client() as client:` which properly called `Client.__aenter__()` (initializing the gRPC channel via `ready()`) and `Client.__aexit__()` (closing the channel via `close()`). The new code passes `self.client()` directly to `Worker(...)` without entering the context manager, so:
   1. `ready()` is never called — the client may not be properly initialized.
   2. `close()` is never called — gRPC channel resources will leak after each test.
   
   This will likely cause integration test failures or resource leaks.

   Suggested fix:
   @asynccontextmanager
   async def worker(
       self, registry: Registry, **kwargs: Unpack[WorkerOptions]
   ) -> AsyncGenerator[Worker, None]:
       async with self.client() as client:
           worker = Worker(client, self.test_name, registry, **kwargs)
           async with worker.run():
               yield worker

2. 💡 Edge Case: Worker tasks may fail silently before yield completes
   Files: cadence/worker/_worker.py:43-50

   In `Worker.run()`, tasks are created via `asyncio.create_task()` and then `yield self` is reached. If a task raises an exception before the caller's `async with` block body executes (e.g., connection failure on first poll), that exception is silently swallowed by `return_exceptions=True` in the `finally` block's `asyncio.gather()`. With the old `TaskGroup` approach, such exceptions would propagate. Consider checking task results after gather and re-raising if any contain non-CancelledError exceptions.

Options

Auto-apply is off → Gitar will not commit updates to this branch.
Display: compact → Showing less information.

Comment with these commands to change:

Auto-apply	Compact
gitar auto-apply:on	gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Gitar}