Harden inbound actor handshakes
Registry probes need short retry deadlines, but applying their one-second budget to every portal and child connection can terminate a valid delayed actor with no client retry path. Give ordinary pre-registration handshakes an independent ten-second deadline. Normalize raw `msgspec.DecodeError` frames to `TransportClosed` so malformed peers cannot cancel the shared IPC nursery with decoder internals. Cover malformed frames and the ordinary-vs-probe timeout distinction. Review: PR #480 (goodboy) https://github.com/goodboy/tractor/pull/480 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))wkt/uds_macos_473
parent
0b63af020e
commit
1a9ce915f3
|
|
@ -4,7 +4,12 @@ High-level `.ipc._server` unit tests.
|
||||||
'''
|
'''
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
import errno
|
import errno
|
||||||
|
from unittest.mock import (
|
||||||
|
AsyncMock,
|
||||||
|
Mock,
|
||||||
|
)
|
||||||
|
|
||||||
|
import msgspec
|
||||||
import pytest
|
import pytest
|
||||||
import trio
|
import trio
|
||||||
from tractor import (
|
from tractor import (
|
||||||
|
|
@ -16,7 +21,10 @@ from tractor._testing.addr import (
|
||||||
get_rando_addr,
|
get_rando_addr,
|
||||||
)
|
)
|
||||||
from tractor._exceptions import TransportClosed
|
from tractor._exceptions import TransportClosed
|
||||||
|
from tractor.ipc._chan import Channel
|
||||||
|
from tractor.ipc import _server
|
||||||
from tractor.ipc._transport import MsgpackTransport
|
from tractor.ipc._transport import MsgpackTransport
|
||||||
|
from tractor.msg.types import Aid
|
||||||
# TODO, use/check-roundtripping with some of these wrapper types?
|
# TODO, use/check-roundtripping with some of these wrapper types?
|
||||||
#
|
#
|
||||||
# from .._addr import Address
|
# from .._addr import Address
|
||||||
|
|
@ -30,8 +38,8 @@ def test_send_normalizes_peer_reset():
|
||||||
'''
|
'''
|
||||||
Normalize Darwin's pre-handshake peer reset as transport closure.
|
Normalize Darwin's pre-handshake peer reset as transport closure.
|
||||||
|
|
||||||
A raw UDS readiness client connects and immediately disconnects.
|
A UDS peer may disconnect before completing the actor handshake.
|
||||||
Darwin reports the server's first handshake write as
|
Darwin can report the server's first handshake write as
|
||||||
`ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing
|
`ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing
|
||||||
that raw error to escape cancels the daemon's shared IPC nursery.
|
that raw error to escape cancels the daemon's shared IPC nursery.
|
||||||
This fake stream reproduces the exact exception chain and proves
|
This fake stream reproduces the exact exception chain and proves
|
||||||
|
|
@ -69,6 +77,90 @@ def test_send_normalizes_peer_reset():
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
|
def test_handshake_normalizes_decode_error():
|
||||||
|
'''
|
||||||
|
Keep malformed pre-handshake frames out of the service nursery.
|
||||||
|
|
||||||
|
A non-msgpack peer can trigger `msgspec.DecodeError` before a
|
||||||
|
remote `Aid` exists. Letting that decoder error escape the inbound
|
||||||
|
handler cancels the actor's shared IPC nursery. This fake channel
|
||||||
|
proves `_do_handshake()` presents only `TransportClosed` upward.
|
||||||
|
|
||||||
|
'''
|
||||||
|
chan = object.__new__(Channel)
|
||||||
|
chan.send = AsyncMock()
|
||||||
|
chan.recv = AsyncMock(
|
||||||
|
side_effect=msgspec.DecodeError('malformed handshake'),
|
||||||
|
)
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
with pytest.raises(TransportClosed) as exc_info:
|
||||||
|
await chan._do_handshake(
|
||||||
|
aid=Aid(
|
||||||
|
name='local',
|
||||||
|
uuid='local-uuid',
|
||||||
|
pid=1234,
|
||||||
|
),
|
||||||
|
timeout=.1,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(
|
||||||
|
exc_info.value.src_exc,
|
||||||
|
msgspec.DecodeError,
|
||||||
|
)
|
||||||
|
|
||||||
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
|
def test_server_uses_independent_handshake_timeout(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Give ordinary actor handshakes a distinct, generous deadline.
|
||||||
|
|
||||||
|
Registry probes use short retries, but ordinary portal and child
|
||||||
|
connections do not retry. Applying the probe's one-second timeout
|
||||||
|
in the server can terminate a valid delayed child and leave its
|
||||||
|
parent blocked in `IPCServer.wait_for_peer()`. This handler fake
|
||||||
|
proves the server uses its separate pre-registration budget.
|
||||||
|
|
||||||
|
'''
|
||||||
|
handshake = AsyncMock(
|
||||||
|
side_effect=TransportClosed(message='stop after assertion'),
|
||||||
|
)
|
||||||
|
chan = Mock(_do_handshake=handshake)
|
||||||
|
actor = Mock(
|
||||||
|
aid=Aid(
|
||||||
|
name='local',
|
||||||
|
uuid='local-uuid',
|
||||||
|
pid=1234,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
Channel,
|
||||||
|
'from_stream',
|
||||||
|
Mock(return_value=chan),
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
_server._state,
|
||||||
|
'current_actor',
|
||||||
|
Mock(return_value=actor),
|
||||||
|
)
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
await _server.handle_stream_from_peer(
|
||||||
|
stream=Mock(),
|
||||||
|
server=Mock(),
|
||||||
|
)
|
||||||
|
|
||||||
|
trio.run(main)
|
||||||
|
handshake.assert_awaited_once_with(
|
||||||
|
aid=actor.aid,
|
||||||
|
timeout=_server._PRE_REG_HANDSHAKE_TIMEOUT,
|
||||||
|
)
|
||||||
|
assert _server._PRE_REG_HANDSHAKE_TIMEOUT == 10
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
'_tpt_proto',
|
'_tpt_proto',
|
||||||
['uds', 'tcp']
|
['uds', 'tcp']
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,7 @@ from typing import (
|
||||||
)
|
)
|
||||||
import warnings
|
import warnings
|
||||||
|
|
||||||
|
import msgspec
|
||||||
import trio
|
import trio
|
||||||
|
|
||||||
from ._types import (
|
from ._types import (
|
||||||
|
|
@ -518,6 +519,7 @@ class Channel:
|
||||||
)
|
)
|
||||||
except (
|
except (
|
||||||
MsgTypeError,
|
MsgTypeError,
|
||||||
|
msgspec.DecodeError,
|
||||||
TypeError,
|
TypeError,
|
||||||
UnicodeDecodeError,
|
UnicodeDecodeError,
|
||||||
trio.TooSlowError,
|
trio.TooSlowError,
|
||||||
|
|
|
||||||
|
|
@ -72,7 +72,7 @@ if TYPE_CHECKING:
|
||||||
|
|
||||||
log = log.get_logger()
|
log = log.get_logger()
|
||||||
|
|
||||||
_PRE_REG_HANDSHAKE_TIMEOUT: float = 1
|
_PRE_REG_HANDSHAKE_TIMEOUT: float = 10
|
||||||
|
|
||||||
|
|
||||||
async def maybe_wait_on_canced_subs(
|
async def maybe_wait_on_canced_subs(
|
||||||
|
|
@ -352,12 +352,9 @@ async def handle_stream_from_peer(
|
||||||
# "kinda-error" that we expect to tolerate during
|
# "kinda-error" that we expect to tolerate during
|
||||||
# discovery-sys related pings, queires, DoS etc.
|
# discovery-sys related pings, queires, DoS etc.
|
||||||
):
|
):
|
||||||
# XXX: This may propagate up from `Channel._aiter_recv()`
|
# `TransportClosed` is expected when a peer disconnects or
|
||||||
# and `MsgpackStream._inter_packets()` on a read from the
|
# fails the initial typed handshake, including foreign clients
|
||||||
# stream particularly when the runtime is first starting up
|
# and probes racing shutdown.
|
||||||
# inside `open_root_actor()` where there is a check for
|
|
||||||
# a bound listener on the registrar addr. the reset will be
|
|
||||||
# because the handshake was never meant took place.
|
|
||||||
log.runtime(
|
log.runtime(
|
||||||
con_status
|
con_status
|
||||||
+
|
+
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue