[mikemikimike]

Summary

Idle worker-loop retirement joined the retired thread without a bound. A task a routine leaves behind can block that thread indefinitely, and the join runs under the loop-pool lock, so one wedged thread stalled the next dispatch and the stop RPC with it. Bound the join by the teardown's own worst case — the shared drain budget plus one per-clear floor for each pool clear the teardown runs, 6.5 s at the defaults — and abandon a thread still running at the bound to the process-exit daemon reap, reported at warning level for idle retirement and at debug under a stop request, whose caller chose the bound it outlived.

Two design decisions carry the correctness argument. First, the join budget's provenance rides a ContextVar set only around a stop's own pool clear: a TTL-driven retirement snapshots the context captured when its expiry timer was armed, so an overlapping stop cannot divert the retirement onto the stop's budget by construction, and the finalizer execution-context contract this relies on is now documented on ResourcePool and pinned by tests. Second, the teardown budgets become constructor configuration on WorkerService, so the tests that exercise these bounds configure them through the API instead of patching private module constants.

LocalWorker sizes its stop-RPC deadline against this contract: twice a finite grace — StopRequest.timeout bounds the in-flight drain and the thread join in sequence — plus a public STOP_RPC_MARGIN covering the retirement worst case and one second of transport.

Closes #426. Closes #447.

Proposed changes

Bound the idle-retirement join

Join the retired worker thread with a timeout of drain_timeout + len(clears) × min_clear_timeout instead of waiting indefinitely. The derivation follows the teardown's two clocks: the drain deadline starts on the worker loop while the join starts on the service loop, so the worst case is the drain budget, one floor for each clear a spent budget can starve, and the floor the first, never-starved clear does not spend — the handoff slack between the clocks and the loop's final stop and close callbacks. A thread still running at the bound is left to the daemon reap and the overrun is logged with its elapsed bound.

Carry the stop budget on a ContextVar

Map StopRequest.timeout once in _stop — negative to an unbounded join, zero to no synchronous join, positive to that many seconds — and set it on _stop_budget only around the stop's own _loop_pool.clear(). The teardown's snapshot keys both the join bound and the report level on the variable's presence: set means a caller chose the bound (debug), unset means idle retirement computed it (warning). Provenance is therefore a property of the finalizer's execution context, not of service-wide state a concurrent stop could latch.

Expose the teardown budgets as constructor configuration

Add drain_timeout, min_clear_timeout, and worker_loop_ttl keyword parameters to WorkerService. The TTL resolves at construction, where the loop pool consumes it; the teardown budgets resolve per teardown, so an unset parameter follows the module default as it stands when a teardown runs. Rename _CLEAR_FLOOR to _MIN_CLEAR_TIMEOUT to match the parameter it defaults.

Size the LocalWorker stop deadline and document the public contract

Set the stop RPC's deadline to 2 × grace + STOP_RPC_MARGIN, publicizing the margin: the retirement worst case plus one second of transport and response overhead, so a stop that lands during an idle retirement — which waits out that retirement's bound regardless of its own timeout — still returns before the deadline forces SIGTERM on a healthy worker. State the contract where callers read it: the tri-state field comment on wire.proto's StopRequest.timeout, the quantified default worst case in WorkerConnection.stop, and the retirement override in both READMEs.

Promise the pool's finalizer execution context

Document on ResourcePool that clear runs finalizers on the calling task and that TTL cleanup runs in the context captured when the timer was armed. The provenance split above rides on this behavior, so it is stated as a contract and pinned by tests rather than left as an implementation accident.

Test cases

The bounds and report levels are mutation-verified: the unbounded-join, budget-latch, unconditional-report, grace-doubling, and margin-term mutants each fail at least one test below.

#	Test Suite	Given	When	Then	Coverage Target
1	TestWorkerService	A worker loop with a blocked daemon thread and an expired idle TTL	Another dispatch arrives	Runs on a new worker thread within the retirement bound	Dispatch availability under a wedged retirement
2	TestWorkerService	A worker loop with a blocked daemon thread	The TTL retirement's join outlives the retirement bound	Warns and leaves the thread alive as a daemon for the process-exit reap	Overrun report with idle provenance
3	TestWorkerService	An idle retirement in flight over a blocked thread	The stop RPC is invoked with a negative timeout	Returns within the retirement bound with the service stopped	Stop return under a concurrent retirement
4	TestWorkerService	A TTL that fires while a stop with a positive grace is still draining in-flight work	The retirement's teardown joins the blocked thread	Joins on the idle bound and warns despite the concurrent stop	Join-budget provenance under stop/TTL overlap
5	TestWorkerService	A warm worker loop with a blocked daemon thread	The stop RPC is invoked with a positive timeout	Returns within twice the timeout and reports the join overrun at debug	Two-wait budget with stop provenance
6	TestWorkerService	A warm worker loop with a blocked daemon thread	The stop RPC is invoked with a zero timeout	Returns within transport slack without joining the thread	Zero-timeout fast path
7	TestWorkerService	A warm worker loop with a blocked daemon thread	The stop RPC is invoked with a negative timeout	Does not return until the thread is released	Unbounded join on request
8	TestResourcePool	A cached entry and a finalizer that reads a context variable	The variable is set around a call to clear	The finalizer observes the caller's value	Finalizer context contract for clear
9	TestResourcePool	An entry released while a context variable is set, arming the expiry timer	The TTL elapses after the variable is reset	The finalizer observes the arm-time value	Finalizer context contract for TTL cleanup
10	TestLocalWorker	Any finite positive grace period	stop() is called	Bounds the RPC by twice the grace plus a margin that itself exceeds the worst-case retirement	Stop-deadline sizing with a draw-independent margin pin

[conradbzura]

Hey @mikemikimike, thanks for the contribution! I re-targeted your branch against master since it's a bugfix that could have been released as a patch to v0.15, but I thought better of it after seeing that it depends on a few unreleased changes on main. It'll go up with v0.16. I'm reviewing it now.

[conradbzura]

This rewrite brings the test in line with the rest of the file and uses the fixture from the helper suggestion:

• A Given-When-Then docstring and Arrange-Act-Assert comments, which every other test here carries.
• A name that says what's asserted: the next dispatch runs on a new worker loop.
• The private timing constants keep their patches, with the same justification comment the TTL test above uses.
• The warning check asserts the level as well as the message, under a caplog scoped to the service's logger.
• elapsed is checked against the patched bound plus _GRPC_SLACK rather than a bare 1.0, which a bound three times too large would still pass.

Suggested change

	@pytest.mark.asyncio
	async def test_dispatch_should_bound_idle_retirement_when_worker_thread_is_blocked(
	self, grpc_aio_stub, mocker: MockerFixture, mock_worker_proxy_cache, caplog
	):
	"""An idle worker retirement must not hold the pool lock indefinitely.
	Given a worker loop whose thread blocks after a completed dispatch,
	When its idle TTL expires and another request arrives,
	Then the second request gets a new worker within the retirement
	budget and the blocked daemon thread is reported.
	"""
	ttl = 0.8
	mocker.patch.object(service_module, "_WORKER_LOOP_TTL", ttl)
	mocker.patch.object(service_module, "_DRAIN_TIMEOUT", 0.2)
	mocker.patch.object(service_module, "_CLEAR_FLOOR", 0.05)
	key = uuid4().hex
	service = WorkerService()
	loop = asyncio.get_running_loop()
	block_started = threading.Event()
	release_block = threading.Event()
	first_thread: threading.Thread | None = None
	try:
	async with grpc_aio_stub(servicer=service) as stub:
	first = stub.dispatch()
	await first.write(
	protocol.Request(
	task=make_task(
	_idle_retirement_worker_loop_probe, args=(key,)
	).to_protobuf()
	)
	)
	await first.done_writing()
	first_responses = [response async for response in first]
	assert first_responses[0].HasField("ack")
	first_thread_name = cloudpickle.loads(first_responses[1].result.dump)
	first_thread = next(
	thread
	for thread in threading.enumerate()
	if thread.name == first_thread_name
	)
	worker_loop = _idle_retirement_probe_loops[key]
	def block_worker_thread():
	block_started.set()
	release_block.wait()
	worker_loop.call_soon_threadsafe(block_worker_thread)
	assert await loop.run_in_executor(None, block_started.wait, 5)
	await asyncio.sleep(ttl + 0.1)
	started = loop.time()
	async with asyncio.timeout(2.0):
	second = stub.dispatch()
	await second.write(
	protocol.Request(
	task=make_task(_worker_loop_identity_probe).to_protobuf()
	)
	)
	await second.done_writing()
	second_responses = [response async for response in second]
	elapsed = loop.time() - started
	assert second_responses[0].HasField("ack")
	_, _, second_thread_name = cloudpickle.loads(second_responses[1].result.dump)
	assert second_thread_name != first_thread_name
	assert elapsed < 1.0
	assert any(
	"Worker-loop thread did not stop" in record.getMessage()
	for record in caplog.records
	)
	finally:
	release_block.set()
	_idle_retirement_probe_loops.pop(key, None)
	if first_thread is not None:
	await loop.run_in_executor(None, first_thread.join, 5)
	assert not first_thread.is_alive()
	@pytest.mark.asyncio
	async def test_dispatch_should_start_on_new_worker_loop_when_retired_thread_blocked(
	self,
	grpc_aio_stub,
	mocker: MockerFixture,
	mock_worker_proxy_cache,
	worker_thread_block,
	caplog,
	):
	"""Test `WorkerService.dispatch` is not held up by a blocked retirement.
	Given:
	A worker loop whose daemon thread is blocked by a task a
	routine left behind, and whose idle TTL has expired
	When:
	Another dispatch arrives
	Then:
	It should run on a new worker thread within the retirement
	bound and warn that the blocked thread was left to the
	daemon reap
	"""
	# Arrange — no public knob exposes the idle TTL, the drain budget
	# or the per-clear floor, so patch them to keep the retirement
	# bound short: the drain budget plus one floor for each of the
	# proxy, subscriber and channel clears.
	mocker.patch.object(service_module, "_WORKER_LOOP_TTL", 0.8)
	mocker.patch.object(service_module, "_DRAIN_TIMEOUT", 0.2)
	mocker.patch.object(service_module, "_CLEAR_FLOOR", 0.05)
	retirement_bound = 0.2 + 3 * 0.05
	key, block = worker_thread_block
	loop = asyncio.get_running_loop()
	blocker = make_task(_leave_worker_thread_blocker, args=(key,))
	probe = make_task(_worker_loop_identity_probe)
	async with grpc_aio_stub(servicer=WorkerService()) as stub:
	first = stub.dispatch()
	await first.write(protocol.Request(task=blocker.to_protobuf()))
	await first.done_writing()
	_, first_result = [r async for r in first]
	first_thread_name = cloudpickle.loads(first_result.result.dump)
	block.armed.set()
	assert await loop.run_in_executor(None, block.entered.wait, 5)
	await asyncio.sleep(0.8 + 0.1)
	# Act
	with caplog.at_level(logging.WARNING, logger="wool.runtime.worker.service"):
	started = loop.time()
	async with asyncio.timeout(5):
	second = stub.dispatch()
	await second.write(protocol.Request(task=probe.to_protobuf()))
	await second.done_writing()
	_, second_result = [r async for r in second]
	elapsed = loop.time() - started
	# Assert
	_, _, second_thread_name = cloudpickle.loads(second_result.result.dump)
	assert second_thread_name != first_thread_name
	assert elapsed < retirement_bound + _GRPC_SLACK
	assert any(
	record.levelno == logging.WARNING
	and "Worker-loop thread did not stop" in record.getMessage()
	for record in caplog.records
	)

[conradbzura]

Most of this landed — the docstring, the phase comments, the rename, and the worker_thread_block fixture are all in. Two pieces of the thread are still open, and verifying the first turned up an arithmetic error worth fixing with it.

The arrange block is now copied across all three tests, and its literal is wrong. Each test repeats the _DRAIN_TIMEOUT/_CLEAR_FLOOR patch pair, the "no public knob" comment, and retirement_bound = 0.2 + 3 * 0.05 — a literal re-derivation of the production expression _DRAIN_TIMEOUT + len(clears) * _CLEAR_FLOOR. The comment counts "one floor for each of the proxy, subscriber and channel clears", but no subscriber pool is installed in these tests (__subscriber_pool__ defaults to None), so the finalizer runs two clears, not three: the service's own warning logs the bound it enforced as 0.300s where the comment derives 0.35. The tests pass because 0.35 is a looser envelope, which means the "within the retirement bound" assertions are 0.05 s slacker than the bound the service actually enforces. One fixture gives the values and the formula a single home and makes the count true by construction, by depending on the pool fixtures it counts:

@pytest.fixture
def retirement_bound(
    mocker: MockerFixture, mock_worker_proxy_cache, mock_subscriber_pool
) -> float:
    """Patch the teardown budgets and return the idle-retirement bound."""
    # No public knob exposes the drain budget or the per-clear floor, so
    # patch them to keep the retirement bound short. The pool fixtures
    # are dependencies so the clear count is true by construction: the
    # finalizer runs one clear each for the proxy, subscriber and
    # channel pools only when the first two are installed. The idle TTL
    # stays per-test: only the retirement tests want the loop to retire.
    drain, floor, clears = 0.2, 0.05, 3
    mocker.patch.object(service_module, "_DRAIN_TIMEOUT", drain)
    mocker.patch.object(service_module, "_CLEAR_FLOOR", floor)
    return drain + clears * floor

Each test then requests retirement_bound in place of mock_worker_proxy_cache and its own patch pair, comment, and literal — and the assertions tighten to the bound the service computes.

Nothing asserts the daemon-reap half of the warning. The dispatch test now scopes caplog correctly, but it bundles three behaviours (new thread, bound, warning), and no assertion anywhere shows the abandoned thread is daemonic and still alive — that half of #426's expected behavior is inferred from the log message alone. Splitting the warning into its own test pins both halves and lets the dispatch test assert exactly what its name claims:

@pytest.mark.asyncio
async def test_dispatch_should_warn_when_retired_thread_outlives_bound(
    self,
    grpc_aio_stub,
    mocker: MockerFixture,
    worker_thread_block,
    retirement_bound,
    caplog,
):
    """Test `WorkerService.dispatch` reports an abandoned worker thread.

    Given:
        A worker loop whose daemon thread is blocked by a task a
        routine left behind, and whose idle TTL has expired
    When:
        Another dispatch arrives after the retirement bound elapses
    Then:
        It should warn that the thread was left to the daemon reap
        while that thread is still alive and daemonic
    """
    # Arrange
    mocker.patch.object(service_module, "_WORKER_LOOP_TTL", 0.8)
    key, block = worker_thread_block
    loop = asyncio.get_running_loop()
    service = WorkerService()
    blocker = make_task(_leave_worker_thread_blocker, args=(key,))
    probe = make_task(_worker_loop_identity_probe)

    async with grpc_aio_stub(servicer=service) as stub:
        first = stub.dispatch()
        await first.write(protocol.Request(task=blocker.to_protobuf()))
        await first.done_writing()
        _ = [response async for response in first]
        block.armed.set()
        assert await loop.run_in_executor(None, block.entered.wait, 5)
        await asyncio.sleep(0.8 + 0.1)

        # Act
        with caplog.at_level(logging.WARNING, logger="wool.runtime.worker.service"):
            second = stub.dispatch()
            await second.write(protocol.Request(task=probe.to_protobuf()))
            await second.done_writing()
            _ = [response async for response in second]

    # Assert
    assert block.thread is not None
    assert block.thread.daemon and block.thread.is_alive()
    assert any(
        record.levelno == logging.WARNING
        and "Worker-loop thread did not stop" in record.getMessage()
        for record in caplog.records
    )

The existing dispatch test then drops its caplog arm (the fixture, the at_level wrapper, and the warning assertion) and keeps the two assertions its name announces: the new thread identity and the elapsed bound.

[conradbzura]

Partially addressed — the bound's arithmetic is now correct; the mechanism chosen to correct it introduces a defect of its own.

Resolved: each test now derives retirement_bound once, with the true clear count of 2, so the asserted bound coincides with the 0.300 s bound the service enforces.

What remains is closable in one edit. The derivation obtains its expected value from the implementation under test — service_module._idle_retirement_join_timeout(2) — so the assertion is no longer an independent oracle: any change to the production formula propagates into the expectation it is checked against (a mutant returning _DRAIN_TIMEOUT * 10 loosens the asserted bound in the same step), leaving the hard wait_for literals as the only discriminating constraint. Deriving the bound from the values the test itself patches restores independence, and correcting the Arrange comments completes the edit: they still attribute one floor to each of "the proxy, subscriber and channel clears", but under these fixtures only the proxy and channel clears are registered (__subscriber_pool__ defaults to None), so the comment contradicts the 2 on the following line.

        # Arrange — no public knob exposes the idle TTL, the drain budget
        # or the per-clear floor, so patch them to keep the retirement
        # bound short: the drain budget plus one floor for each of the
        # proxy and channel clears — no subscriber pool is installed
        # under these fixtures.
        mocker.patch.object(service_module, "_WORKER_LOOP_TTL", 0.8)
        mocker.patch.object(service_module, "_DRAIN_TIMEOUT", 0.2)
        mocker.patch.object(service_module, "_CLEAR_FLOOR", 0.05)
        retirement_bound = 0.2 + 2 * 0.05

Also outstanding from this thread: the patch triple and its five-line comment now appear three times, and the dispatch-and-arm sequence (write the blocker, drain the responses, arm, entered.wait) four times; a retirement_bound fixture and a _wedge_worker_thread(stub, key, block, loop) helper collapse both.

The same extraction is the occasion to remove the group's fixed waits, and the motivation is structural. A fixed sleep costs its worst case on every execution, independent of outcome, and must be sized for the slowest environment that will ever run it, so its margin is monotonically non-decreasing over the suite's life; each additional test written in this shape adds that worst case to every run. A wait on an observable event costs the actual latency of the condition, confines its timeout to the failure path, and cannot fail by being too tight — which removes the incentive to pad. Two changes suffice, measured on a prototype. First, synchronize on the retirement's own report rather than sleeping past its scheduled time:

        retired = asyncio.Event()

        class _RetirementSignal(logging.Handler):
            def emit(self, record):
                if "still running" in record.getMessage():
                    loop.call_soon_threadsafe(retired.set)

with the handler installed on logging.getLogger("wool.runtime.worker.service") for the test's duration and await retired.wait() under asyncio.timeout(5) replacing await asyncio.sleep(0.8 + 0.1). Second, reduce the patched budgets by an order of magnitude (_WORKER_LOOP_TTL 0.2, _DRAIN_TIMEOUT 0.05, _CLEAR_FLOOR 0.01): a wedged thread consumes its full join budget deterministically, so the budget's size contributes no flake margin on these paths. Measured effect: the dispatch test's body falls from ~1.1 s to ~0.3 s, and the test still fails — through the 5 s backstop — when the join is reverted to unbounded.

Two constraints bound the technique, one measured and one structural. The wedge must be armed only after the first dispatch's response has drained — arming earlier wedges the loop under the in-flight response and deadlocks the test (observed) — so the TTL must remain an order of magnitude above the arm-to-entered latency; 0.2 s satisfies this, and the current 0.8 s purchases a margin the event wait no longer needs. And two waits admit no event substitute: the stop-during-retirement test must position its stop inside the retirement window, whose only observable fires at the window's end, so it remains time-positioned — enlarging the patched _DRAIN_TIMEOUT relative to the positioning sleep makes it simultaneously cheaper and less erodible — and the negative-timeout test's still-pending proof must, by construction, outwait the bound a defective implementation would apply, so it scales down with the budgets rather than vanishing.

Separately, the warning assertion still resides in the dispatch test, and no assertion establishes that the abandoned thread is daemonic and alive (assert block.thread.daemon and block.thread.is_alive()) before release.

[mikemikimike]

Validation update for head 4f99b3a:

• The three focused worker-loop regression tests covering the review feedback pass.
• The broader validation limitations are reproducible on base 77a88e4 and are not introduced by this PR. Installing wool[dev] on Windows stops because uvloop 0.23.0 does not support Windows. The non-integration suite reaches tests/cicd/test_bump_version.py, whose .sh helper Windows cannot launch (WinError 193).
• The full worker test module stalls at the same synchronous-dispatch test on both base and PR; that test passes when run alone on both. The PR integration run stalled at a channel-pool test, which passes individually on both base and PR; the base integration run also stalled later at a separate test. The full integration result remains inconclusive.
• Ruff check and format pass on the changed Python files, and git diff --check passes. Project-wide Ruff findings (two lint and two formatting findings) and three Pyright errors are in unchanged files and reproduce on base.
• GitHub currently reports no checks for this head.

[conradbzura]

@mikemikimike I had a little bit more feedback. It's looking good.

[conradbzura]

The provenance mechanism has no distinguishing test. Substituting self._stopping.is_set() for _stop_teardown.get() in this expression — which inverts the provenance fix exactly — passes all 131 tests in this file and all 1,283 across tests/runtime/worker and tests/runtime/test_resourcepool.py, measured twice against the current branch. Under that mutant, a TTL-initiated finalizer whose first step executes after a stop(timeout=-1) has set _stopping performs thread.join(None) on the wedged thread while holding the pool lock — the precise hang the provenance fix exists to preclude.

A direct test is obstructed by scheduling. The divergence window is one event-loop turn wide: _stopping.set() must occur between the TTL timer firing and the finalize task's first step. The race-compressed construction — ensure_future(stub.stop(StopRequest(timeout=-1))) issued in the turn the TTL expires — measured 0/5 window hits against the mutant, because the finalize task's first step always precedes the RPC's path to _stopping.set(). Advancing the stop inverts the hazard: when the stop's clear() acquires the pool lock first, the finalizer correctly runs under the stop's unbounded budget and the test hangs on a correct implementation.

The enforceable alternative is contractual. The fix depends on two ResourcePool properties — clear() executes finalizers inline on the calling task, and a TTL finalizer executes in the context captured at timer arm — neither documented nor tested on the pool. Pinning both with direct pool tests converts the service-side dependence into a checked contract (see the companion comment at _stop's teardown scope), and a one-line comment beside the overlap test, naming the untestable one-turn window, records the residue.

[conradbzura]

The correctness of this selection rests on two ResourcePool properties the pool exhibits but nowhere states: clear() executes finalizers inline on the calling task, which is what makes this token visible at the budget snapshot; and a TTL-driven finalizer executes in the context captured when its timer was armed, which is never a stop context. The finalizer parameter's documentation specifies lock-holding and exception semantics only. Were ResourcePool to execute clear-sweep finalizers on newly spawned tasks with explicit contexts, or to capture contexts at release time, this flag would read False during a stop — silently bounding a negative-timeout stop — and no test of the pool would register the change.

The property should be promoted to a contract: state on ResourcePool (on the finalizer parameter, or on clear) that a finalizer executes in the context of the operation that triggers it, and pin the statement with two pool tests — a ContextVar set around clear() is visible inside the finalizer, and a TTL-fired finalizer observes the arm-time context. These two tests also constitute the practical enforcement for the provenance gap noted at the budget snapshot, whose direct race admits no deterministic test.

[conradbzura]

No assertion establishes that the stop overlapped the in-flight retirement. The overlap window is 0.2 s wide — the TTL cleanup begins at t≈0.8 s and its bounded join completes near t≈1.1 s, while the stop is issued after sleep(0.9) plus RPC latency — so under load the stop can arrive after the retirement has completed, and the test degrades silently into a stop-after-retirement scenario: still green, no longer exercising the overlap. The kill direction is unaffected — an unbounded retirement never completes, so the wait_for(5) still fails on a regression — which makes this an erosion of intent rather than a correctness hole. Two measures preserve the intent: widen the window (a larger patched _DRAIN_TIMEOUT buys width directly), and assert a witness of the overlap — the retirement's "still running" report not yet emitted at the instant the stop is issued is the direct one.

[conradbzura]

@mikemikimike A few more test-related items and I think we're good to merge. I'll make a documentation pass once the remaining feedback is addressed.