From 5b16130c633acbde069f33f14a06a3f81996d34e Mon Sep 17 00:00:00 2001 From: Conrad Date: Tue, 11 Aug 2026 14:07:20 -0400 Subject: [PATCH] refactor: Rename the worker idle_time surface to idle The surface spelled one concept three ways: the wire RPC and the service handler said idle, the caller-facing method said idle_time, and the message said IdleTime. WorkerConnection.idle now mirrors the RPC it wraps, and the unit stays explicit where it matters: the message's seconds field and the method docstring. The RPC name itself is untouched, so the gRPC method path and the seconds field tag are identical and the change is wire-compatible with v0.14.0-rc0 workers. The surface has only shipped in that pre-release, so renaming the Python API before v0.14.0 final avoids a deprecation cycle. IdleUnavailable keeps its name. --- wool/proto/wire.proto | 4 +- wool/src/wool/protocol/__init__.py | 4 +- wool/src/wool/protocol/_wire.py | 4 +- wool/src/wool/runtime/worker/connection.py | 17 ++++---- wool/src/wool/runtime/worker/service.py | 6 +-- wool/tests/integration/test_worker_idle.py | 26 +++++------ wool/tests/protocol/test_wire.py | 22 +++++----- wool/tests/runtime/worker/test_connection.py | 45 +++++++++----------- 8 files changed, 62 insertions(+), 66 deletions(-) diff --git a/wool/proto/wire.proto b/wool/proto/wire.proto index 61dd2b93..5a61eb70 100644 --- a/wool/proto/wire.proto +++ b/wool/proto/wire.proto @@ -104,7 +104,7 @@ message Nack { service Worker { rpc dispatch (stream Request) returns (stream Response); rpc stop (StopRequest) returns (Void); - rpc idle (Void) returns (IdleTime); + rpc idle (Void) returns (Idle); } message Request { @@ -136,7 +136,7 @@ message StopRequest { float timeout = 1; } -message IdleTime { +message Idle { // Seconds the worker has been continuously idle: time since the // in-flight task set last emptied (startup counts as empty). // 0 while any task is in flight; measured on a monotonic clock. diff --git a/wool/src/wool/protocol/__init__.py b/wool/src/wool/protocol/__init__.py index 81ec5b00..1d569c71 100644 --- a/wool/src/wool/protocol/__init__.py +++ b/wool/src/wool/protocol/__init__.py @@ -13,7 +13,7 @@ from wool.protocol._wire import ChainManifest as ChainManifest from wool.protocol._wire import ChannelOptions as ChannelOptions from wool.protocol._wire import ContextVar as ContextVar -from wool.protocol._wire import IdleTime as IdleTime +from wool.protocol._wire import Idle as Idle from wool.protocol._wire import Message as Message from wool.protocol._wire import Nack as Nack from wool.protocol._wire import Request as Request @@ -36,7 +36,7 @@ "ChainManifest", "ChannelOptions", "ContextVar", - "IdleTime", + "Idle", "Message", "Nack", "Request", diff --git a/wool/src/wool/protocol/_wire.py b/wool/src/wool/protocol/_wire.py index d9874b42..456ef0d5 100644 --- a/wool/src/wool/protocol/_wire.py +++ b/wool/src/wool/protocol/_wire.py @@ -14,7 +14,7 @@ from wool.protocol.wire_pb2 import ChainManifest from wool.protocol.wire_pb2 import ChannelOptions from wool.protocol.wire_pb2 import ContextVar - from wool.protocol.wire_pb2 import IdleTime + from wool.protocol.wire_pb2 import Idle from wool.protocol.wire_pb2 import Message from wool.protocol.wire_pb2 import Nack from wool.protocol.wire_pb2 import Request @@ -58,7 +58,7 @@ def __call__(servicer, server) -> None: ... "ChainManifest", "ChannelOptions", "ContextVar", - "IdleTime", + "Idle", "Message", "Nack", "Request", diff --git a/wool/src/wool/runtime/worker/connection.py b/wool/src/wool/runtime/worker/connection.py index e92cbb8b..683523db 100644 --- a/wool/src/wool/runtime/worker/connection.py +++ b/wool/src/wool/runtime/worker/connection.py @@ -206,13 +206,12 @@ class IdleUnavailable(WoolError): """Raised when a worker does not implement the idle RPC. A worker that predates the idle capability answers the idle RPC - with gRPC ``UNIMPLEMENTED``; `WorkerConnection.idle_time` - translates that into this typed signal so a polling client can - detect an old worker and treat idle reporting as unavailable, - distinct from a transient hiccup or an unhealthy peer. It descends - from `wool.WoolError` — not `RpcError` — because an absent - capability is not an RPC-health fault, so ``except RpcError`` does - not catch it. + with gRPC ``UNIMPLEMENTED``; `WorkerConnection.idle` translates + that into this typed signal so a polling client can detect an old + worker and treat idle reporting as unavailable, distinct from a + transient hiccup or an unhealthy peer. It descends from + `wool.WoolError` — not `RpcError` — because an absent capability is + not an RPC-health fault, so ``except RpcError`` does not catch it. """ @@ -220,7 +219,7 @@ class IdleUnavailable(WoolError): class WorkerConnection: """Direct single-worker control surface over a pooled gRPC channel. - Exposes `dispatch` (task execution), `idle_time` (poll the worker's + Exposes `dispatch` (task execution), `idle` (poll the worker's continuous idle duration), and `stop` (shut the remote worker down). ``close`` is distinct: it releases this connection's local pooled channel, whereas `stop` terminates the remote worker. @@ -487,7 +486,7 @@ async def close(self): if self._uds_key is not None: await _channel_pool.expire(self._uds_key) - async def idle_time(self, *, timeout: float | None = None) -> float: + async def idle(self, *, timeout: float | None = None) -> float: """Query how long the remote worker has been continuously idle. Returns the worker's reported continuous idle duration; see diff --git a/wool/src/wool/runtime/worker/service.py b/wool/src/wool/runtime/worker/service.py index fac3cbe5..8e2a9269 100644 --- a/wool/src/wool/runtime/worker/service.py +++ b/wool/src/wool/runtime/worker/service.py @@ -582,7 +582,7 @@ async def idle( self, request: protocol.Void, context: ServicerContext | None, - ) -> protocol.IdleTime: + ) -> protocol.Idle: """Report how long the worker has been continuously idle. Idle is the number of seconds since the in-flight task set @@ -600,13 +600,13 @@ async def idle( :param context: The `grpc.aio.ServicerContext` for this request. :returns: - An `IdleTime` carrying the continuous idle duration in + An `Idle` carrying the continuous idle duration in seconds. """ seconds = ( 0.0 if self._idle_since is None else time.monotonic() - self._idle_since ) - return protocol.IdleTime(seconds=seconds) + return protocol.Idle(seconds=seconds) @staticmethod def _create_worker_loop( diff --git a/wool/tests/integration/test_worker_idle.py b/wool/tests/integration/test_worker_idle.py index 16f584b9..b2f69a87 100644 --- a/wool/tests/integration/test_worker_idle.py +++ b/wool/tests/integration/test_worker_idle.py @@ -109,7 +109,7 @@ class TestWorkerIdleReporting: "cred", [CredentialType.INSECURE, CredentialType.MTLS, CredentialType.ONE_WAY], ) - async def test_idle_time_should_accrue_from_startup_over_the_real_wire( + async def test_idle_should_accrue_from_startup_over_the_real_wire( self, credentials_map, retry_grpc_internal, cred ): """Test idle accrues from startup, over each transport. @@ -133,9 +133,9 @@ async def body(): conn = connect() # Act - first = await conn.idle_time() + first = await conn.idle() await asyncio.sleep(0.1) - second = await conn.idle_time() + second = await conn.idle() # Assert assert first >= 0.0 @@ -145,7 +145,7 @@ async def body(): await retry_grpc_internal(body) @pytest.mark.asyncio - async def test_idle_time_should_report_zero_while_a_task_is_in_flight( + async def test_idle_should_report_zero_while_a_task_is_in_flight( self, credentials_map, retry_grpc_internal, tmp_path ): """Test idle is zero while a dispatched routine runs on the worker. @@ -177,7 +177,7 @@ async def body(): ) # Act & assert - assert await conn.idle_time() == 0.0 + assert await conn.idle() == 0.0 finally: dispatch.cancel() with contextlib.suppress(BaseException): @@ -186,7 +186,7 @@ async def body(): await retry_grpc_internal(body) @pytest.mark.asyncio - async def test_idle_time_should_reset_after_the_in_flight_set_drains( + async def test_idle_should_reset_after_the_in_flight_set_drains( self, credentials_map, retry_grpc_internal, tmp_path ): """Test idle resets once the worker's in-flight set drains. @@ -210,7 +210,7 @@ async def body(): ): conn = connect() await asyncio.sleep(0.2) - before = await conn.idle_time() + before = await conn.idle() # Act sentinel = tmp_path / "drain.txt" @@ -219,10 +219,10 @@ async def body(): # Wait for the docket-drain to be reflected (idle > 0), # then confirm it counts from the drain, not from startup. async def _reset(): - return await conn.idle_time() > 0.0 + return await conn.idle() > 0.0 await _poll_coro(_reset) - after = await conn.idle_time() + after = await conn.idle() # Assert assert before > 0.0 @@ -231,7 +231,7 @@ async def _reset(): await retry_grpc_internal(body) @pytest.mark.asyncio - async def test_idle_time_should_raise_idle_unavailable_for_a_legacy_worker( + async def test_idle_should_raise_idle_unavailable_for_a_legacy_worker( self, retry_grpc_internal ): """Test idle surfaces IdleUnavailable against a legacy worker. @@ -257,7 +257,7 @@ async def body(): try: # Act & assert with pytest.raises(IdleUnavailable) as exc_info: - await conn.idle_time() + await conn.idle() assert not isinstance(exc_info.value, RpcError) finally: await conn.close() @@ -287,7 +287,7 @@ async def body(): # Arrange async with _bare_worker(credentials_map[cred]) as (worker, connect): conn = connect() - assert await conn.idle_time() >= 0.0 + assert await conn.idle() >= 0.0 # Act await conn.stop() @@ -295,7 +295,7 @@ async def body(): # Assert async def _unreachable(): try: - await conn.idle_time() + await conn.idle() return False except TransientRpcError: return True diff --git a/wool/tests/protocol/test_wire.py b/wool/tests/protocol/test_wire.py index 0423e876..c2f62b0d 100644 --- a/wool/tests/protocol/test_wire.py +++ b/wool/tests/protocol/test_wire.py @@ -8,7 +8,7 @@ EXPECTED_MESSAGE_EXPORTS = [ "Ack", "ChainManifest", - "IdleTime", + "Idle", "Message", "Nack", "Request", @@ -273,38 +273,38 @@ def test_nack_without_exception(self): parsed.ParseFromString(nack.SerializeToString()) assert parsed.HasField("exception") is False - def test_idle_time_fields(self): - """Test IdleTime carries the idle duration in seconds. + def test_idle_fields(self): + """Test Idle carries the idle duration in seconds. Given: A seconds value, and the default construction. When: - An IdleTime message is constructed with and without a value. + An Idle message is constructed with and without a value. Then: The seconds field should hold the value and default to 0.0. """ # Arrange, act, & assert - assert protocol.IdleTime(seconds=42.5).seconds == 42.5 - assert protocol.IdleTime().seconds == 0.0 + assert protocol.Idle(seconds=42.5).seconds == 42.5 + assert protocol.Idle().seconds == 0.0 @settings(max_examples=100) @given(seconds=st.floats(width=64, allow_nan=False, allow_infinity=False)) - def test_idle_time_roundtrip(self, seconds): - """Test IdleTime.seconds survives the wire-format round-trip. + def test_idle_roundtrip(self, seconds): + """Test Idle.seconds survives the wire-format round-trip. Given: Any finite double value for the idle duration. When: - An IdleTime message is serialized and re-parsed. + An Idle message is serialized and re-parsed. Then: The seconds field should equal the original exactly — a proto double round-trips losslessly. """ # Arrange - message = protocol.IdleTime(seconds=seconds) + message = protocol.Idle(seconds=seconds) # Act - parsed = protocol.IdleTime() + parsed = protocol.Idle() parsed.ParseFromString(message.SerializeToString()) # Assert diff --git a/wool/tests/runtime/worker/test_connection.py b/wool/tests/runtime/worker/test_connection.py index 897076c4..da82a7e9 100644 --- a/wool/tests/runtime/worker/test_connection.py +++ b/wool/tests/runtime/worker/test_connection.py @@ -1473,14 +1473,13 @@ async def test_dispatch_should_drain_inflight_stream_when_superseded( await connection.close() @pytest.mark.asyncio - async def test_idle_time_should_return_seconds_when_worker_responds( + async def test_idle_should_return_seconds_when_worker_responds( self, mocker: MockerFixture ): """Test idle returns the worker's reported idle seconds. Given: - A connection whose worker answers the idle RPC with an - IdleTime + A connection whose worker answers the idle RPC with an Idle When: idle is awaited Then: @@ -1488,19 +1487,19 @@ async def test_idle_time_should_return_seconds_when_worker_responds( """ # Arrange mock_stub = mocker.MagicMock() - mock_stub.idle = mocker.AsyncMock(return_value=protocol.IdleTime(seconds=42.5)) + mock_stub.idle = mocker.AsyncMock(return_value=protocol.Idle(seconds=42.5)) mocker.patch.object(protocol, "WorkerStub", return_value=mock_stub) connection = WorkerConnection("localhost:50051") # Act - result = await connection.idle_time() + result = await connection.idle() # Assert assert result == 42.5 mock_stub.idle.assert_awaited_once() @pytest.mark.asyncio - async def test_idle_time_should_raise_idle_unavailable_when_unimplemented( + async def test_idle_should_raise_idle_unavailable_when_unimplemented( self, mocker: MockerFixture ): """Test idle surfaces IdleUnavailable for a worker without the RPC. @@ -1530,7 +1529,7 @@ def details(self): # Act & assert with pytest.raises(IdleUnavailable): - await connection.idle_time() + await connection.idle() @pytest.mark.asyncio @settings( @@ -1544,7 +1543,7 @@ def details(self): ), details=st.one_of(st.none(), st.just(""), st.text()), ) - async def test_idle_time_should_map_status_code_to_typed_exception( + async def test_idle_should_map_status_code_to_typed_exception( self, mocker: MockerFixture, code: grpc.StatusCode, details: str | None ): """Test idle maps every gRPC status code to the right exception. @@ -1574,7 +1573,7 @@ async def test_idle_time_should_map_status_code_to_typed_exception( # Act with pytest.raises(Exception) as exc_info: - await connection.idle_time() + await connection.idle() # Assert raised = exc_info.value @@ -1598,23 +1597,23 @@ async def test_idle_time_should_map_status_code_to_typed_exception( suppress_health_check=[HealthCheck.function_scoped_fixture], ) @given(timeout=st.floats(max_value=0.0, allow_nan=False, allow_infinity=False)) - async def test_idle_time_should_reject_non_positive_timeout( + async def test_idle_should_reject_non_positive_timeout( self, mocker: MockerFixture, timeout: float ): - """Test idle_time rejects a non-positive timeout with ValueError. + """Test idle rejects a non-positive timeout with ValueError. Given: A connection whose worker answers the idle RPC, and any non-positive timeout When: - idle_time is awaited with that timeout + idle is awaited with that timeout Then: It should raise ValueError without calling the stub, matching dispatch's timeout validation. """ # Arrange mock_stub = mocker.MagicMock() - mock_stub.idle = mocker.AsyncMock(return_value=protocol.IdleTime(seconds=1.0)) + mock_stub.idle = mocker.AsyncMock(return_value=protocol.Idle(seconds=1.0)) mocker.patch.object(protocol, "WorkerStub", return_value=mock_stub) connection = WorkerConnection("localhost:50051") # Fresh channel per example (see idle-map note above). @@ -1622,7 +1621,7 @@ async def test_idle_time_should_reject_non_positive_timeout( # Act & assert with pytest.raises(ValueError): - await connection.idle_time(timeout=timeout) + await connection.idle(timeout=timeout) mock_stub.idle.assert_not_awaited() @pytest.mark.asyncio @@ -1642,30 +1641,30 @@ async def test_idle_time_should_reject_non_positive_timeout( ), ) ) - async def test_idle_time_should_forward_positive_timeout( + async def test_idle_should_forward_positive_timeout( self, mocker: MockerFixture, timeout: float | None ): - """Test idle_time forwards None or a positive timeout to the stub. + """Test idle forwards None or a positive timeout to the stub. Given: A connection whose worker answers the idle RPC, and None or any positive timeout When: - idle_time is awaited with that timeout + idle is awaited with that timeout Then: It should call the stub with a Void request and that timeout as the gRPC deadline. """ # Arrange mock_stub = mocker.MagicMock() - mock_stub.idle = mocker.AsyncMock(return_value=protocol.IdleTime(seconds=1.0)) + mock_stub.idle = mocker.AsyncMock(return_value=protocol.Idle(seconds=1.0)) mocker.patch.object(protocol, "WorkerStub", return_value=mock_stub) connection = WorkerConnection("localhost:50051") # Fresh channel per example (see idle-map note above). await clear_channel_pool() # Act - result = await connection.idle_time(timeout=timeout) + result = await connection.idle(timeout=timeout) # Assert assert result == 1.0 @@ -1681,7 +1680,7 @@ async def test_idle_time_should_forward_positive_timeout( suppress_health_check=[HealthCheck.function_scoped_fixture], ) @given(seconds=st.floats(width=64, allow_nan=False, allow_infinity=False)) - async def test_idle_time_should_return_reported_seconds_verbatim( + async def test_idle_should_return_reported_seconds_verbatim( self, mocker: MockerFixture, seconds: float ): """Test idle returns the worker's reported seconds losslessly. @@ -1696,16 +1695,14 @@ async def test_idle_time_should_return_reported_seconds_verbatim( """ # Arrange mock_stub = mocker.MagicMock() - mock_stub.idle = mocker.AsyncMock( - return_value=protocol.IdleTime(seconds=seconds) - ) + mock_stub.idle = mocker.AsyncMock(return_value=protocol.Idle(seconds=seconds)) mocker.patch.object(protocol, "WorkerStub", return_value=mock_stub) connection = WorkerConnection("localhost:50051") # Fresh channel per example (see idle-map note above). await clear_channel_pool() # Act - result = await connection.idle_time() + result = await connection.idle() # Assert assert result == seconds