diff --git a/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.md b/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.md new file mode 100644 index 00000000..a7dde9c8 --- /dev/null +++ b/ai/prompt-io/opencode/20260916_ib_disconnect_cleanup.md @@ -0,0 +1,73 @@ +--- +model: openai/gpt-6-astra +service: opencode +scope: code +substantive: true +git_ref: 0d95a9e6 +raw_file: 20260916_ib_disconnect_cleanup.raw.md +--- + +## Prompt + +Fix runtime failures exposed when stopping IB Gateway: idle chart +consumers miss disconnect notification, infected actors retain closed +service nurseries, late SIGINT raises in cancel_soon, and creator shm +segments survive actor exit. Develop on piker_pin in isolation and do +not change the live piker setup. + +## Response summary + +Notify unfinished parent contexts on the exact failed IPC channel, +preserve delivered outcomes and original transport causes, and record +buffered Return outcomes before subsequent EOF. Schedule synchronous +actor cancellation in the root nursery rather than a possibly closed +service nursery, deduplicating repeated requests. + +## Files changed + +- tractor/runtime/_rpc.py: propagate transport closure to ctx owners. +- tractor/_context.py: preserve transport cause and buffered outcomes. +- tractor/runtime/_runtime.py: root-owned cancellation task. +- tests/test_ipc_eof.py: active/idle peers and graceful EOF regressions. +- tests/test_cancel_soon.py: closed-service cancellation and actual + creator-owned shm unlinking under native/infected runtimes. + +## Verification + +These are session observations, not authenticated passes for the +individual commit-plan boundary trees. The shared provenance is carried +by the final cancellation boundary and describes both runtime fixes. +Exact-tree checks remain pending in the commit plan. + +Baseline EOF cases: three failures and two passes. Fixed TCP selection: +34 passes; UDS EOF cases: five passes. Combined cancellation and EOF +cases: nine passes. Extended actual-SHM cancellation test: four passes; +original method injected into fresh actors: four expected failures. +No user processes or shm names were touched; no installations or +commits were made. Full Gateway/SIGINT/debugger sequence remains for +integration validation; automatic IB reconnect is not implemented. + +## Human contributions + +The human supplied incident traces, selected piker_pin, identified +dependent piker compatibility branches, and explicitly required worktree +isolation and preserving the live setup. Acceptance remains pending. + +## Local review adjustment + +The human challenged treating a locally observed transport loss like +a remote task failure/cancellation and requested readability changes. +Removed invented peer attribution from Context._canceller, added +assertions for absent cancellation provenance, restored the Return +warning, expanded predicates, and documented test ordering and fault +semantics. The proposed EOF-driven scope cancellation remains in place +only pending a separate contract decision; test success is not policy +acceptance. Piker-level explicit disconnect supervision is an alternative +to changing the runtime default. Issues #400, #368, and #22 were cited +by the human for the design discussion, not fetched in this pass. + +Review verification: five EOF tests and two service-nursery cases passed. +Seven inter-peer cancellation cases could not start because disabling +automatic reaping removes their track_orphaned_uds_per_test fixture. +That coverage remains pending; no reaper was enabled against live state. +The existing advanced-fault expectations also remain to be reconciled. diff --git a/tests/test_ipc_eof.py b/tests/test_ipc_eof.py new file mode 100644 index 00000000..8de8f6e6 --- /dev/null +++ b/tests/test_ipc_eof.py @@ -0,0 +1,322 @@ +''' +Exercise the proposed discovered-peer EOF notification policy. + +''' +from unittest.mock import Mock + +import pytest +import trio +from trio.testing import wait_all_tasks_blocked + +import tractor +from tractor._exceptions import TransportClosed +from tractor.msg.types import Return + + +_streams: dict[str, tractor.Context] = {} + + +@tractor.context +async def stream_peer( + ctx: tractor.Context, +) -> None: + ''' + Publish a live stream and echo until its caller finishes. + + ''' + _streams[ctx.cid] = ctx + await ctx.started() + async with ctx.open_stream() as stream: + await stream.send('ready') + async for value in stream: + if value == 'finish': + break + await stream.send(value) + + +async def send_peer_eof(cid: str) -> None: + ''' + Close only this test channel's outgoing transport half. + + This helper sends no Stop, Return, or Error. Keeping the read + half open avoids a local fd-close error crashing the peer actor + and confusing transport propagation with supervision. + + ''' + ctx = _streams[cid] + await ctx.chan._transport.stream.send_eof() + + +@tractor.context +async def consume_peer( + owner: tractor.Context, + wait_on: str, + reg_addr, +) -> bool: + ''' + Exercise a sibling connection outside child supervision. + + ''' + with trio.fail_after(10): + async with ( + # Discovery creates a sibling channel, independent of + # the root actor's control connection to eof_peer. + tractor.find_actor( + 'eof_peer', + registry_addrs=[reg_addr], + ) as peer, + # This second peer is a control: the fault on peer's + # channel must not terminate unrelated communications. + tractor.find_actor( + 'healthy_peer', + registry_addrs=[reg_addr], + ) as healthy, + healthy.open_context(stream_peer) as (good, _), + good.open_stream() as good_stream, + trio.open_nursery() as tn, + ): + # owner.chan carries the root's orchestration messages; + # losing peer.channel must not lose the result channel. + assert peer.channel is not owner.chan + # Confirm the healthy context is live before injecting + # the fault, not merely connect to it after teardown. + assert await good_stream.receive() == 'ready' + errors: list[TransportClosed] = [] + + # Set only once BOTH affected context owners observe + # TransportClosed; one notified owner is insufficient. + finished = trio.Event() + + # Model an application checkpoint unrelated to IPC. + # Never release it for fault cases: only the proposed + # context-scope cancellation may interrupt that wait. + # Completed contexts are explicitly released after EOF. + paused = trio.Event() + + # A local report hook confirms process_messages handled + # EOF before we inspect a completed context's outcome. + eof_seen = trio.Event() + graceful = wait_on in {'returned', 'buffered'} + report = TransportClosed.report_n_maybe_raise + + def report_eof(exc, *args, **kwargs): + # During this gate only the injected peer channel + # closes; healthy/control teardown happens later. + eof_seen.set() + return report(exc, *args, **kwargs) + + # This sibling is a fresh test actor; observe the real + # handler without changing its exception policy. + TransportClosed.report_n_maybe_raise = report_eof + + async def consume( + task_status=trio.TASK_STATUS_IGNORED, + ): + try: + async with ( + peer.open_context(stream_peer) as (ctx, _), + ctx.open_stream() as stream, + ): + assert await stream.receive() == 'ready' + if graceful: + # Observe Context._deliver_msg receipt, + # without requiring the owner to drain + # the final result from its feeder. + returned = trio.Event() + deliver = ctx._deliver_msg + + async def record_return(msg): + result = await deliver(msg) + if isinstance(msg, Return): + returned.set() + return result + + ctx._deliver_msg = record_return + # Ask the remote context to finish while + # leaving its shared actor channel open. + await stream.send('finish') + await returned.wait() + if wait_on == 'returned': + assert ( + await ctx.wait_for_result() + ) is None + # Export the cid only after stream startup + # and any graceful Return delivery. The root + # uses it to select the faulted connection. + task_status.started(ctx) + if graceful: + await paused.wait() + assert ctx._remote_error is None + elif wait_on == 'receive': + await stream.receive() + else: + await paused.wait() + if not graceful: + pytest.fail('resumed without EOF') + except TransportClosed as exc: + assert not graceful + assert exc is ctx._remote_error + assert exc.__cause__ is exc._src_exc + assert 'closed' in exc.message + # This is a local transport observation, not + # evidence that a peer requested cancellation. + assert ctx.canceller is None + assert not ctx.cancel_called + assert not ctx.cancel_acked + errors.append(exc) + if len(errors) == 2: + finished.set() + + # Both contexts share one discovered channel but own + # separate local scopes. EOF must notify both owners. + first = await tn.start(consume) + second = await tn.start(consume) + assert first.chan is second.chan + # Neither readiness handshake alone proves the owner is + # blocked at the intended checkpoint. Quiescence does. + await wait_all_tasks_blocked() + original_send = peer.channel.send + checking_eof = True + + async def send_after_notification(*args, **kwargs): + # Cleanup may send, but must not discover the EOF. + if checking_eof: + assert first._remote_error is not None + assert second._remote_error is not None + return await original_send(*args, **kwargs) + + if not graceful: + peer.channel.send = send_after_notification + # Release the root's fault injection only after both + # owners are waiting and the no-send-detection guard is + # installed. The root uses a separate control channel. + await owner.started(first.cid) + try: + with trio.fail_after(2): + await eof_seen.wait() + if graceful: + paused.set() + else: + await finished.wait() + finally: + checking_eof = False + assert len(errors) == (0 if graceful else 2) + assert paused.is_set() is graceful + assert good._remote_error is None + # Prove the unrelated peer still serves application + # traffic after EOF handling, not just that it has no + # recorded error yet. + await good_stream.send('still alive') + assert await good_stream.receive() == 'still alive' + return True + + +@pytest.mark.parametrize( + 'wait_on', + [ + # Unfinished context blocked in its receive API: wake it + # without requiring an outbound write to discover EOF. + 'receive', + + # Unfinished context blocked on an application Event: + # exercise the proposed interruption of non-IPC awaits. + 'event', + + # Final Return already consumed: preserve success when the + # connection closes after its remote task has completed. + 'returned', + + # Final Return received but not drained: preserve the same + # success even though the owner has not requested its result. + 'buffered', + ], +) +def test_discovered_peer_eof( + reg_addr, + wait_on: str, +) -> None: + ''' + Exercise the proposed EOF-driven cancellation of context owners. + + Previously process_messages logged TransportClosed and returned, + leaving parent Context scopes alive until a later outbound send. + Two owners share a discovered channel, independently of the + ActorNursery's child-control channel. Both receive an initial + stream item, then block in receive() or on an unset Event. + Nursery.start() and wait_all_tasks_blocked() establish that + ordering before a control RPC half-closes the peer transport. + + Under the proposed policy both owners must raise TransportClosed + before the deadline without sending to discover the failure. + The Event case deliberately asserts interruption of a non-IPC + checkpoint; whether to adopt that contract remains under review. + An unrelated peer stream must still echo afterward and close + gracefully. The original EOF cause must survive unwinding. + + Graceful variants deliver Return before EOF, either draining its + value or leaving it buffered while the owner pauses. Observing + delivery and the EOF report synchronizes those races without a + timing sleep. Neither completed owner may be cancelled by EOF. + + ''' + async def main(): + async with ( + tractor.open_nursery( + registry_addrs=[reg_addr], + ) as an, + ): + # Retain a root-to-peer portal exclusively for injecting + # EOF. Its transport is not the sibling channel tested. + control = await an.start_actor( + 'eof_peer', + enable_modules=[__name__], + ) + await an.start_actor( + 'healthy_peer', + enable_modules=[__name__], + ) + # A separate consumer ensures failure cannot propagate + # merely through the root's child-supervision nursery. + consumer = await an.start_actor( + 'consumer', + enable_modules=[__name__], + ) + async with consumer.open_context( + consume_peer, + wait_on=wait_on, + reg_addr=reg_addr, + ) as (ctx, cid): + # consume_peer exports cid after both owners are + # blocked; only then break their shared connection. + await control.run(send_peer_eof, cid=cid) + assert await ctx.wait_for_result() is True + + await an.cancel() + + trio.run(main) + + +def test_disconnect_cause_survives_unwind() -> None: + ''' + Preserve the transport cause when cancelling a context owner. + + Context result/stream receivers can supply their cancellation as + from_src_exc while unwinding. That must not replace the original + socket error in TransportClosed.__cause__. Exercise the actual + raise path with a reset cause and a distinct unwinding exception; + both exception identity and the original cause must survive. + + ''' + ctx = Mock() + ctx._is_self_cancelled.return_value = False + cause = ConnectionResetError('test peer reset') + error = TransportClosed('test disconnect', src_exc=cause) + with pytest.raises(TransportClosed) as raised: + tractor.Context._maybe_raise_remote_err( + ctx, + error, + from_src_exc=RuntimeError('owner unwinding'), + ) + assert raised.value is error + assert error.__cause__ is cause + assert error.src_exc is cause diff --git a/tractor/_context.py b/tractor/_context.py index 57f24ca0..75c20960 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -828,6 +828,17 @@ class Context: f'{error}' ) + elif isinstance(error, TransportClosed): + # TransportClosed is observed locally, not an Error msg + # identifying a cancellation requester or failed task. + # Preserve Context._canceller rather than inventing a + # peer-originated cancellation from connection loss. + cid: str = self.cid + log.warning( + f'Locally observed transport fault for ctx {cid!r}\n' + f'{error!r}' + ) + elif isinstance(error, MsgTypeError): msgerr = True self._canceller = error.src_uid @@ -855,7 +866,11 @@ class Context: f'{error!r}' ) - if self._canceller is None: + if ( + self._canceller is None + and + not isinstance(error, TransportClosed) + ): log.error('Ctx has no canceller set!?') cs: trio.CancelScope = self._scope @@ -878,7 +893,8 @@ class Context: # `.open_context()` block. if ( cs - and not cs.cancel_called + and + not cs.cancel_called # XXX this is an expected cancel request response # message and we **don't need to raise it** in the @@ -1310,7 +1326,13 @@ class Context: # runtime frames from the tb explicitly? # https://docs.python.org/3/reference/simple_stmts.html#the-raise-statement # https://stackoverflow.com/a/24752607 - if from_src_exc is not False: + # A transport failure already carries the disconnect cause. + # Do not replace it with cancellation used to wake the owner. + if ( + from_src_exc is not False + and + not isinstance(remote_error, TransportClosed) + ): raise remote_error from from_src_exc raise remote_error @@ -1924,13 +1946,16 @@ class Context: stream._stop_msg = msg case Return(): - if not self._outcome_msg: + if self._outcome_msg is Unresolved: log.warning( f'Setting final outcome msg AFTER ' - f'`._rx_chan.send()`??\n' + f'`._send_chan.send_nowait()`??\n' f'\n' f'{msg}' ) + # Record receipt even while the owner is + # idle and has not drained the result yet. + # A subsequent EOF must not replace Return. self._outcome_msg = msg return True diff --git a/tractor/runtime/_rpc.py b/tractor/runtime/_rpc.py index e1e86e5d..70c96955 100644 --- a/tractor/runtime/_rpc.py +++ b/tractor/runtime/_rpc.py @@ -46,6 +46,7 @@ from trio import ( from ..ipc import Channel from .._context import ( Context, + Unresolved, ) from .._exceptions import ( ContextCancelled, @@ -1368,6 +1369,35 @@ async def process_messages( ) except TransportClosed as tc: + # MsgpackTransport._iter_pkts() raises TransportClosed for + # a socket EOF (including an empty frame header), or wraps + # a reset/read error. This is not MsgStream's EndOfChannel + # after a protocol Stop, nor a Context's final Return. + # + # Proposed policy, pending the transport-fault review: + # an unfinished parent Context has lost its connection, + # NOT necessarily its remote task. The call below cancels + # Context._scope even when its owner is blocked in unrelated + # application code. That checkpoint sees trio.Cancelled; + # open_context() subsequently raises TransportClosed while + # unwinding. This intentionally remains distinct from a + # peer's explicit cancellation request/remote Error msg. + # + # Actor._contexts can include multiple connections to one + # peer. Match Context.chan by identity to leave other channels + # alone. Preserve received outcomes, including a Return which + # its owner has not drained, instead of replacing them by EOF. + for ctx in tuple(actor._contexts.values()): + if ( + ctx.chan is chan + and ctx.side == 'parent' + and ctx._scope is not None + and ctx._remote_error is None + and ctx._outcome_msg is Unresolved + and not ctx._final_result_is_set() + ): + ctx._maybe_cancel_and_set_remote_error(tc) + # channels "breaking" (for TCP streams by EOF or 104 # connection-reset) is ok since we don't have a teardown # handshake for them (yet) and instead we simply bail out of