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