diff --git a/ai/prompt-io/opencode/20260919_context_fault_status.md b/ai/prompt-io/opencode/20260919_context_fault_status.md new file mode 100644 index 00000000..ddb320e7 --- /dev/null +++ b/ai/prompt-io/opencode/20260919_context_fault_status.md @@ -0,0 +1,145 @@ +--- +model: openai/gpt-6-astra +service: opencode +scope: code +substantive: true +git_ref: dd758da5 +raw_file: 20260919_context_fault_status.raw.md +--- + +## Prompt and human design + +The human prototyped ContextCancelReason, ContextStatus, and a rename of +the context-owned nursery. They requested implementation after saving +the IB fixes, on a separate downstream branch, with Trio's minimum +version in a separate commit. Their design separates runtime-layer +transport faults from reported remote task failures/cancellations. + +Design references supplied by the human: +https://github.com/goodboy/tractor/issues/400 +https://github.com/goodboy/tractor/issues/22 +https://github.com/goodboy/tractor/issues/368 +https://github.com/goodboy/tractor/issues/388 + +## Response summary + +Context._status is the single store for outcome, cancellation protocol, +and handshake/lifecycle state. On explicit human instruction, the +agent removed the generated typed descriptors and duplicate Context +declarations. Internal accesses now use Context._status directly; +there is no moved-field constructor compatibility layer. Policy and +live resources remain on Context. Status diagnostics render only +declared state. Public properties, wait_for_result(), and the deprecated +result() method remain available. + +ContextCancelReason is shallow-frozen and includes context/actor/peer +identity, an opaque scope token and explicit fault origin. The retained +exception is mutable and may retain traceback resources. Local validation +and boxed remote errors remain distinguishable even with equal peer IDs. +Raw transport loss no longer causes self-cancel classification to read +remote-error-only attributes before checking its exception type. + +## Files changed + +- tractor/_context.py: status extraction, metadata and direct access. +- tractor/_exceptions.py: preserve local validation provenance. +- tractor/runtime/_rpc.py: nursery rename and direct status access. +- tractor/_streaming.py: stream state and result status access. +- tractor/msg/_ops.py: final message drain status access. +- tractor/runtime/_runtime.py: Context cancellation status access. +- tractor/devx/debug/_trace.py: copied nursery rename documentation. +- tractor/devx/debug/_tty_lock.py: copied nursery rename documentation. +- tests/test_context_status.py: twenty reason/state/public API cases. +- tests/test_context_stream_semantics.py: status assertions and the + overrun test double's ContextStatus allocation. +- tests/test_ipc_eof.py: status assertions. +- tests/msg/test_pldrx_limiting.py: status assertions. +- tests/test_inter_peer_cancellation.py: status assertions. + +## Verification + +The initial implementation passed 86 tests: 16 context-status, +65 context-stream semantics and 5 EOF cases. Used the existing original +worktree interpreter with Trio0.34.0 and per-process source overrides; +no installed packages or live services were changed. Automatic reaping +was disabled. Broader cancellation suites needing reaper-provided +fixtures remain unverified. + +After the human-directed simplification, the targeted run passed 91 +tests: 20 context-status, 65 context-stream semantics, 5 EOF, and the +overlapping service-nursery cancellation regression. The first run +found a SimpleNamespace overrun test double still declaring +_stream_opened directly; it now allocates ContextStatus. The complete +targeted run passed after that correction and removal of mk_context's +redundant result initialization. + +A separate run passed all 8 payload-limiting cases. All 7 cases in +test_inter_peer_cancellation.py stopped at fixture setup because +track_orphaned_uds_per_test belongs to the disabled reaper plugin. +No cleanup plugin was enabled to bypass this constraint. + +Both runs used the existing ib_disconnect_cleanup/py313/bin/python, +Trio 0.34.0, and PYTHONPATH ordered as this worktree, that interpreter's +site-packages, then piker/py313 site-packages. TRACTOR_SPAWN_METHOD and +TRACTOR_LOGLEVEL were unset. Pytest arguments, after python -m pytest: + +```text +-p no:tractor._testing._reap +tests/test_context_status.py +tests/test_context_stream_semantics.py +tests/test_ipc_eof.py +tests/test_resource_cache.py::test_moc_service_nursery_cancellation_completes_exit +-q +``` + +Separate process: + +```text +-p no:tractor._testing._reap +tests/msg/test_pldrx_limiting.py +tests/test_inter_peer_cancellation.py +-q +``` + +The source audit accounts for all 70 old production direct accesses: +49 rewritten in _context.py, its redundant mk_context assignment +removed, and 20 rewritten across seven functions in four other files. +All 33 old direct test accesses across five files were addressed: +29 migrated and four compatibility assertions replaced with status +initialization/isolation and public API assertions. The additional +test-double keyword dependency was found by execution and corrected. +Remaining same-named attributes belong to Actor, ActorNursery or +Channel. Added Python lines fit 69 columns; git diff --check passes. + +A private pre-adjustment index tree and the real index SHA256 are +recorded in ignored .claude runtime state for review. The existing +plan-build helper captured the private tree without git add. The +real index and HEAD remain unchanged; all remediation is unstaged. + +## Human contribution accounting + +The prototype classes, reason= experiment, nursery rename, downstream +sequencing, fault-layer distinction, and separate dependency boundary +were human-directed. The human accepted immutable identity metadata +and excluding policies/resources from the status store. The agent added +the actual authoritative store, descriptor delegation, provenance +classification, diagnostics and regressions in the initial output. + +The human subsequently rejected descriptor delegation and its +self-imposed constructor/attribute compatibility tests. They directed +the single-store simplification, the precise 11-field migration scope, +preservation of public APIs and the deprecated result method, fault +origin policy, comments/TODO markers and _scope_tn rename. They also +required preserving their staged prototype and leaving the remediation +unstaged. These are human-directed agent edits, not direct manual +source edits. The original raw generated record is retained unchanged. +Acceptance remains human-owned. + +## Limits + +This does not add new supervision/reconnect APIs, inject Cancelled +subclasses, or selectively uncancel scopes. First cancellation reason +does not necessarily track the latest remote error. copy.copy(Context) +shares the status object, as it also shares other live context resources. +Private-field constructor compatibility and dataclasses.replace() +compatibility assertions were removed on human instruction. diff --git a/ai/prompt-io/opencode/20260919_context_fault_status.raw.md b/ai/prompt-io/opencode/20260919_context_fault_status.raw.md new file mode 100644 index 00000000..9c622c85 --- /dev/null +++ b/ai/prompt-io/opencode/20260919_context_fault_status.raw.md @@ -0,0 +1,18 @@ +--- +model: openai/gpt-6-astra +service: opencode +git_ref: dd758da5 +diff_cmd: git diff HEAD +--- + +> git diff HEAD -- tractor/_context.py tractor/_exceptions.py tractor/runtime/_rpc.py tractor/devx/debug/_trace.py tractor/devx/debug/_tty_lock.py tests/test_context_status.py + +Implement authoritative ContextStatus state with typed descriptor +delegation, immutable cancellation identity/origin metadata, and +status-specific formatting. Retain the human's _scope_tn rename. +Correct transport-error classification and distinguish locally +observed validation errors from boxed remote task failures. + +The final working implementation passed 86 tests in 79.88 seconds +using Trio0.34.0: sixteen new reason/status tests, sixty-five context +stream semantics tests, and five IPC EOF regressions. diff --git a/tests/msg/test_pldrx_limiting.py b/tests/msg/test_pldrx_limiting.py index 24173920..b04111b0 100644 --- a/tests/msg/test_pldrx_limiting.py +++ b/tests/msg/test_pldrx_limiting.py @@ -378,10 +378,10 @@ def test_basic_payload_spec( # await tractor.pause(shield=True) # verify expected remote mte deats - assert ctx._local_error is None + assert ctx._status._local_error is None assert ( mte is - ctx._remote_error is + ctx._status._remote_error is ctx.maybe_error is ctx.outcome ) diff --git a/tests/test_context_stream_semantics.py b/tests/test_context_stream_semantics.py index d20b2143..be67dd0c 100644 --- a/tests/test_context_stream_semantics.py +++ b/tests/test_context_stream_semantics.py @@ -33,6 +33,7 @@ from tractor._exceptions import ( ContextCancelled, TransportClosed, ) +from tractor._context import ContextStatus from tractor.runtime._state import current_ipc_ctx from tractor._testing import ( @@ -152,7 +153,7 @@ def test_overrun_error_send_tolerates_transport_close( repr_caller='test', _in_overrun=False, _actor=SimpleNamespace(aid=local_aid), - _stream_opened=True, + _status=ContextStatus(_stream_opened=True), _allow_overruns=False, ) msg = tractor.msg.Yield( @@ -644,9 +645,9 @@ def test_parent_cancels( assert ( err is ctx.maybe_error - is ctx._remote_error + is ctx._status._remote_error ) - if le := ctx._local_error: + if le := ctx._status._local_error: assert err is le # else: @@ -654,7 +655,7 @@ def test_parent_cancels( # not defined until block closes right? # # await tractor.pause() - # assert ctx._local_error is None + # assert ctx._status._local_error is None # TODO: don't need this right? # if _ctxc: @@ -892,11 +893,11 @@ async def expect_cancelled( assert not ctx._scope.cancelled_caught # should be the RPC cmd request for `._cancel_task()` - assert ctx._cancel_msg + assert ctx._status._cancel_msg # which, has not yet resolved to an error outcome # since this rpc func has not yet exited. assert not ctx.maybe_error - assert not ctx._final_result_is_set() + assert not ctx._status._final_result_is_set() # debug REPL if needed # with trio.CancelScope(shield=True): @@ -1040,8 +1041,8 @@ async def test_parent_exits_ctx_after_child_enters_stream( # rxed ctxc response from far end assert ctx.cancel_acked assert ( - ctx._remote_error - is ctx._local_error + ctx._status._remote_error + is ctx._status._local_error is ctx.maybe_error is ctx.outcome ) diff --git a/tests/test_inter_peer_cancellation.py b/tests/test_inter_peer_cancellation.py index 05cea5dc..3a43b1e7 100644 --- a/tests/test_inter_peer_cancellation.py +++ b/tests/test_inter_peer_cancellation.py @@ -144,7 +144,8 @@ async def error_before_started( # is part of a local error and we shouldn't raise # ``ContextCancelled`` **iff** we **were not** the side # of the context to initiate it, i.e. - # ``Context._cancel_called`` should **NOT** have been + # ``ContextStatus._cancel_called`` should **NOT** + # have been # set. The special logic to handle this case is now # inside ``Context._maybe_raise_from_remote_msg()`` XD await peer_ctx.cancel() @@ -270,7 +271,7 @@ async def stream_from_peer( # print("TRYING TO ENTER PAUSSE!!!") # await tractor.pause(shield=True) - re: ContextCancelled = peer_ctx._remote_error + re: ContextCancelled = peer_ctx._status._remote_error # XXX YES XXX, remote error should be unpacked only once! assert ( @@ -280,7 +281,7 @@ async def stream_from_peer( is ctxc is - peer_ctx._local_error + peer_ctx._status._local_error ) # NOTE: these errors should all match! # ------ - ------ @@ -293,7 +294,7 @@ async def stream_from_peer( # the same remote error bubbling out above vs what was # already unpacked and set inside `Context. assert ( - peer_ctx._remote_error.msgdata + peer_ctx._status._remote_error.msgdata == ctxc.msgdata ) @@ -301,10 +302,12 @@ async def stream_from_peer( # the above larger assert makes sense to also always be true! # XXX YES XXX, bc should be exact same msg instances - assert peer_ctx._remote_error._ipc_msg is ctxc._ipc_msg + assert ( + peer_ctx._status._remote_error._ipc_msg is ctxc._ipc_msg + ) # XXX NO XXX, bc new one always created for property accesss - assert peer_ctx._remote_error.ipc_msg != ctxc.ipc_msg + assert peer_ctx._status._remote_error.ipc_msg != ctxc.ipc_msg # the peer ctx is the canceller even though it's canceller # is the "canceller" XD @@ -577,13 +580,19 @@ def test_peer_canceller( # a `trio.Cancelled` due to a local # `._scope.cancel()` call. assert not sleeper_ctx._scope.cancelled_caught - # assert sleeper_ctx._local_error is _loc_err - # assert sleeper_ctx._local_error is _loc_err + # assert ( + # sleeper_ctx._status._local_error + # is _loc_err + # ) + # assert ( + # sleeper_ctx._status._local_error + # is _loc_err + # ) assert not ( loc_err is sleeper_ctx.maybe_error is sleeper_ctx.outcome - is sleeper_ctx._remote_error + is sleeper_ctx._status._remote_error ) # NOTE: this root actor task should have @@ -595,8 +604,8 @@ def test_peer_canceller( # cancelled all opened contexts except the # sleeper which is obvi by the "canceller" # peer. - re = ctx._remote_error - le = ctx._local_error + re = ctx._status._remote_error + le = ctx._status._local_error assert ctx.cancel_called @@ -666,13 +675,17 @@ def test_peer_canceller( assert ( final_err is loc_err is sleeper_ctx.maybe_error - is sleeper_ctx._remote_error + is sleeper_ctx._status._remote_error ) for name, ctx in ctxs.items(): - re: BaseException|None = ctx._remote_error - le: BaseException|None = ctx._local_error + re: BaseException|None = ( + ctx._status._remote_error + ) + le: BaseException|None = ( + ctx._status._local_error + ) err = ctx.maybe_error out = ctx.outcome @@ -693,10 +706,10 @@ def test_peer_canceller( if ctx is sleeper_ctx: assert re assert ( - ctx._remote_error + ctx._status._remote_error is ctx.maybe_error is ctx.outcome - is ctx._local_error + is ctx._status._local_error ) assert not ctx.cancel_called @@ -715,7 +728,7 @@ def test_peer_canceller( canceller_ctx, ): - assert not ctx._remote_error + assert not ctx._status._remote_error # neither of the `caller/canceller_ctx` should # have called `ctx.cancel()` bc the @@ -750,7 +763,7 @@ def test_peer_canceller( assert not ctx._scope.cancelled_caught # elif ctx is canceller_ctx: - # assert not ctx._remote_error + # assert not ctx._status._remote_error # XXX NOTE XXX: ONLY the canceller # will get a self-cancelled outcome @@ -794,11 +807,11 @@ def test_peer_canceller( # set in both outcomes including the case where # ctx-cancel handling itself errors. assert not sleeper_ctx._scope.cancelled_caught - assert _loc_err is sleeper_ctx._local_error + assert _loc_err is sleeper_ctx._status._local_error assert ( sleeper_ctx.outcome is sleeper_ctx.maybe_error - is sleeper_ctx._remote_error + is sleeper_ctx._status._remote_error ) raise # always to ensure teardown diff --git a/tests/test_ipc_eof.py b/tests/test_ipc_eof.py index 8de8f6e6..2cc30ca7 100644 --- a/tests/test_ipc_eof.py +++ b/tests/test_ipc_eof.py @@ -146,7 +146,7 @@ async def consume_peer( task_status.started(ctx) if graceful: await paused.wait() - assert ctx._remote_error is None + assert ctx._status._remote_error is None elif wait_on == 'receive': await stream.receive() else: @@ -155,7 +155,7 @@ async def consume_peer( pytest.fail('resumed without EOF') except TransportClosed as exc: assert not graceful - assert exc is ctx._remote_error + assert exc is ctx._status._remote_error assert exc.__cause__ is exc._src_exc assert 'closed' in exc.message # This is a local transport observation, not @@ -181,8 +181,8 @@ async def consume_peer( 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 + assert first._status._remote_error is not None + assert second._status._remote_error is not None return await original_send(*args, **kwargs) if not graceful: @@ -202,7 +202,7 @@ async def consume_peer( checking_eof = False assert len(errors) == (0 if graceful else 2) assert paused.is_set() is graceful - assert good._remote_error is None + assert good._status._remote_error is None # Prove the unrelated peer still serves application # traffic after EOF handling, not just that it has no # recorded error yet. diff --git a/tractor/_context.py b/tractor/_context.py index 75c20960..f41744f5 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -54,6 +54,7 @@ from typing import ( Any, AsyncGenerator, Callable, + Literal, Type, TypeAlias, TYPE_CHECKING, @@ -119,12 +120,135 @@ log = get_logger() class Unresolved: ''' - Placeholder value for `Context._result` until + Placeholder value for `ContextStatus._result` until a final return value or raised error is resolved. ''' +class ContextCancelReason(pretty_struct.Struct, frozen=True): + ''' + Encapsulate an IPC ctx's task cancellation cause. + + Metadata is shallow-frozen, not a wire message. `error` retains + a mutable exception (and possibly its traceback/resources) for + the lifetime of this reason. It is the first cancelling fault, + not necessarily the latest `ContextStatus._remote_error`. + + `scope_token` is an opaque identity token, never a scope ref. + Compare the public `trio.Cancelled.reason` to distinguish this + scope's cancellation from an ancestor's cancellation. + + ''' + message: str + error: BaseException + cid: str + actor_uid: tuple[str, str] + peer_uid: tuple[str, str] + scope_token: object + origin: Literal[ + 'local_transport', + 'remote_task', + 'remote_cancel', + 'local_protocol', + 'local_error', + ] + + # TODO, `.__str__()` impl for if/when trio-core operates + # on a `CancelScope.cancel(reason=)`-argument presuming it + # is always `str`. + + +class ContextStatus(pretty_struct.Struct): + ''' + Authoritative per-context outcome and IPC dialog state. + + Live resources and policy remain on `Context`. Internal callers + access these fields through `Context._status`; only these + declared status fields appear in this diagnostic repr. + + ''' + _outcome_msg: Return|Error|ContextCancelled|type[Unresolved] = ( + Unresolved + ) + + # on a clean exit there should be a final value + # delivered from the far end "child" task, so + # this value is only set on one side. + # _result: Any | int = None + _result: PayloadT|type[Unresolved] = Unresolved + + # if the local "parent" task errors this value is always set + # to the error that was captured in the + # `Portal.open_context().__aexit__()` teardown block OR, in + # 2 special cases when an (maybe) expected remote error + # arrives that we purposely swallow silently: + # - `ContextCancelled` with `.canceller` set to our uid: + # a self-cancel, + # - `RemoteActorError[StreamOverrun]` which was caught during + # a self-cancellation teardown msg drain. + _local_error: BaseException|None = None + + # if the either side gets an error from the other + # this value is set to that error unpacked from an + # IPC msg. + _remote_error: BaseException|None = None + + # only set if an actor-local task called `.cancel()` + # did WE request cancel of the far end? + _cancel_called: bool = False + + # TODO: do we even need this? we can assume that if we're + # cancelled that the other side is as well, so maybe we should + # instead just have a `.canceller` pulled from the + # `ContextCancelled`? + _canceller: tuple[str, str]|None = None + + # NOTE: we try to ensure assignment of a "cancel msg" since + # there's always going to be an "underlying reason" that any + # context was closed due to either a remote side error or + # a call to `.cancel()` which triggers `ContextCancelled`. + _cancel_msg: str|dict|MsgType|None = None + + # init and streaming state + _started_called: bool = False + _started_msg: MsgType|None = None + _started_pld: Any = None + _stream_opened: bool = False + + def _final_result_is_set(self) -> bool: + ''' + Report a resolved result, not merely a buffered Return. + + ''' + return self._result is not Unresolved + + def pformat( + self, + field_indent: int = 2, + indent: int = 0, + ) -> str: + ''' + Display status without changing global private-field rules. + + Accept the same indentation arguments as `Struct.pformat`. + `repr(status)` and `str(status)` use this override; an + explicit `pretty_struct.pformat(status)` still uses the + global formatter's private-field filtering. + + ''' + prefix: str = ' ' * (indent + field_indent) + closing: str = ' ' * indent + fields: str = '\n'.join( + f'{prefix}{name}={getattr(self, name)!r},' + for name in self.__struct_fields__ + ) + return f'ContextStatus(\n{fields}\n{closing})' + + __repr__ = pformat + __str__ = pformat + + # TODO: make this a .msg.types.Struct! # -[ ] ideally we can freeze it # -[ ] let's us do field diffing nicely in tests Bo @@ -220,6 +344,14 @@ class Context: # full "namespace-path" to target RPC function _nsf: NamespacePath + _status: ContextStatus = field( + default_factory=ContextStatus, + init=False, + ) + # Each Context owns one local scope lifetime. Reasons retain + # this identity marker, not the scope or its task tree. + _scope_token: object = field(default_factory=object, init=False) + # the "invocation type" of the far end task-entry-point # function, normally matching a logic block inside # `._runtime.invoke()`. @@ -252,45 +384,6 @@ class Context: # a drain loop? # _res_scope: trio.CancelScope|None = None - _outcome_msg: Return|Error|ContextCancelled = Unresolved - - # on a clean exit there should be a final value - # delivered from the far end "child" task, so - # this value is only set on one side. - # _result: Any | int = None - _result: PayloadT|Unresolved = Unresolved - - # if the local "parent" task errors this value is always set - # to the error that was captured in the - # `Portal.open_context().__aexit__()` teardown block OR, in - # 2 special cases when an (maybe) expected remote error - # arrives that we purposely swallow silently: - # - `ContextCancelled` with `.canceller` set to our uid: - # a self-cancel, - # - `RemoteActorError[StreamOverrun]` which was caught during - # a self-cancellation teardown msg drain. - _local_error: BaseException|None = None - - # if the either side gets an error from the other - # this value is set to that error unpacked from an - # IPC msg. - _remote_error: BaseException|None = None - - # only set if an actor-local task called `.cancel()` - _cancel_called: bool = False # did WE request cancel of the far end? - - # TODO: do we even need this? we can assume that if we're - # cancelled that the other side is as well, so maybe we should - # instead just have a `.canceller` pulled from the - # `ContextCancelled`? - _canceller: tuple[str, str]|None = None - - # NOTE: we try to ensure assignment of a "cancel msg" since - # there's always going to be an "underlying reason" that any - # context was closed due to either a remote side error or - # a call to `.cancel()` which triggers `ContextCancelled`. - _cancel_msg: str|dict|None = None - # NOTE: this state-var is used by the runtime to determine if the # `pdbp` REPL is allowed to engage on contexts terminated via # a `ContextCancelled` due to a call to `.cancel()` triggering @@ -303,11 +396,6 @@ class Context: # actors from being able to acquire the debugger lock. _enter_debugger_on_cancel: bool = True - # init and streaming state - _started_called: bool = False - _started_msg: MsgType|None = None - _started_pld: Any = None - _stream_opened: bool = False _stream: MsgStream|None = None # the parent-task's calling-fn's frame-info, the frame above @@ -331,9 +419,7 @@ class Context: # starting a task inside this nursery that simply raised the # boxed exception. NOW, it's used for spawning overrun queuing # tasks when `.allow_overruns == True` !!! - _scope_nursery: trio.Nursery|None = None - # ^-TODO-^ change name? - # -> `._scope_tn` "scope task nursery" + _scope_tn: trio.Nursery|None = None # streaming overrun state tracking _in_overrun: bool = False @@ -395,10 +481,10 @@ class Context: f' |_state: {self.repr_state!r}\n' f' outcome{ds}{outcome_str}\n' - f' result{ds}{self._result}\n' + f' result{ds}{self._status._result}\n' f' cancel_called{ds}{self.cancel_called}\n' f' cancel_acked{ds}{self.cancel_acked}\n' - f' canceller{ds}{self._canceller}\n' + f' canceller{ds}{self._status._canceller}\n' # TODO: any other fields? # f' maybe_error{ds}{self.maybe_error}\n' @@ -406,7 +492,7 @@ class Context: # f' cancelled_caught={self.cancelled_caught}\n' # -[ ] remove this ^ right? - # f' _remote_error={self._remote_error} + # f' _remote_error={self._status._remote_error} ) if extra_fields: for key, val in extra_fields.items(): @@ -451,7 +537,7 @@ class Context: the `Portal.open_context()` block. ''' - return self._cancel_called + return self._status._cancel_called # XXX, to debug who frickin sets it.. # @cancel_called.setter @@ -464,7 +550,7 @@ class Context: # from .devx import pause_from_sync # pause_from_sync() - # self._cancel_called = val + # self._status._cancel_called = val # TODO, use the `Actor.aid: Aid` instead! @property @@ -475,7 +561,7 @@ class Context: (side of the) context to also be cancelled. ''' - if canc := self._canceller: + if canc := self._status._canceller: return tuple(canc) return None @@ -486,18 +572,18 @@ class Context: ) -> bool: - if not self._cancel_called: + if not self._status._cancel_called: return False re: BaseException|None = ( remote_error or - self._remote_error + self._status._remote_error ) # XXX we only report "this context" as self-cancelled # once we've received a ctxc from our direct-peer task # (aka we're `.cancel_acked`). - if not re: + if not isinstance(re, ContextCancelled): return False if from_uid := re.src_uid: @@ -577,7 +663,7 @@ class Context: # and instead likely we received a remote side # # cancellation that was raised inside `.wait_for_result()` # or ( - # (se := self._local_error) + # (se := self._status._local_error) # and se is re # ) ) @@ -709,6 +795,8 @@ class Context: def _maybe_cancel_and_set_remote_error( self, error: BaseException, + *, + local_protocol: bool = False, # TODO: manual toggle for cases where we wouldn't normally # mark ourselves cancelled but want to? @@ -785,7 +873,7 @@ class Context: # result processing. # TODO: never do this right? - # if self._remote_error: + # if self._status._remote_error: # return # XXX: denote and set the remote side's error so that @@ -798,13 +886,16 @@ class Context: f'=> {self.side!r}: {self._actor.aid.reprol()}\n\n' f'{error!r}' ) - self._remote_error: BaseException = error + self._status._remote_error: BaseException = error msgerr: bool = False + message: str|None = None + origin = 'local_error' # self-cancel (ack) or, # peer propagated remote cancellation. if isinstance(error, ContextCancelled): + origin = 'remote_cancel' # NOTE in the case error is a ctxc the canceller will # either be another peer or us. in the case where it's us # we mark ourself as the canceller of ourselves (a ctx @@ -818,37 +909,53 @@ class Context: # !TODO, switching to `Actor.aid` here! if (canc := error.canceller) == self._actor.aid.uid: whom: str = 'us' - self._canceller = canc + self._status._canceller = canc else: whom = 'a remote peer (not us)' - self._canceller = error.src_uid + self._status._canceller = error.src_uid - log.cancel( - f'IPC context was cancelled by {whom}!\n\n' - f'{error}' + message: str = ( + f'IPC context was cancelled by {whom}!\n' + f'\n' + f'{error!r}' ) + log.cancel(message) elif isinstance(error, TransportClosed): + origin = 'local_transport' # 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. + # Preserve ContextStatus._canceller rather than inventing + # a peer-originated cancellation from connection loss. cid: str = self.cid - log.warning( + message: str = ( f'Locally observed transport fault for ctx {cid!r}\n' f'{error!r}' ) + log.warning(message) elif isinstance(error, MsgTypeError): + # An emulated Error can name the peer as src_uid even + # when PldRx detected the invalid payload locally. + origin = ( + 'local_protocol' if local_protocol + else 'remote_task' + ) msgerr = True - self._canceller = error.src_uid - log.error( - f'IPC dialog error due to msg-type caused by {self.peer_side!r} side\n\n' + self._status._canceller = error.src_uid + peer_side: str = self.peer_side + message: str = ( + f'IPC dialog error due to msg-type caused by ' + f'{peer_side!r} side\n' + f'\n' f'{error}\n' f'{pformat(self)}\n' ) + log.error(message) else: + if isinstance(error, RemoteActorError): + origin = 'remote_task' # always record the cancelling actor's uid since its # cancellation state is linked and we want to know # which process was the cause / requester of the @@ -859,7 +966,7 @@ class Context: None, ) # we mark the source actor as our canceller - self._canceller = maybe_error_src_uid + self._status._canceller = maybe_error_src_uid log.error( f'Remote context error:\n\n' # f'{pformat(self)}\n' @@ -867,7 +974,7 @@ class Context: ) if ( - self._canceller is None + self._status._canceller is None and not isinstance(error, TransportClosed) ): @@ -915,16 +1022,35 @@ class Context: ): message: str = ( f'NOT Cancelling `Context._scope` since,\n' - f'Context._cancel_on_msgerr = {self._cancel_on_msgerr}\n\n' + f'Context._cancel_on_msgerr = {self._cancel_on_msgerr}\n' + f'\n' f'AND we got a msg-type-error!\n' - f'{error}\n' + f'{error!r}\n' ) else: # TODO: it'd sure be handy to inject our own # `trio.Cancelled` subtype here ;) # https://github.com/goodboy/tractor/issues/368 - message: str = 'Cancelling `Context._scope` !\n\n' - cs.cancel() + submsg: str = ( + message + or + f'{error!r}\n' + ) + message: str = ( + 'Cancelling `Context._scope` due to,\n' + f'{submsg}' + ) + cs.cancel( + reason=ContextCancelReason( + message=message, + error=error, + cid=self.cid, + actor_uid=self._actor.aid.uid, + peer_uid=self.chan.aid.uid, + scope_token=self._scope_token, + origin=origin, + ) + ) # TODO, explicit condition for OoB (self-)cancellation? # - we called `Portal.cancel_actor()` from this actor @@ -957,7 +1083,17 @@ class Context: # `._task` directly. IF the owner task is however # blocking elsewhere we need to interrupt it **now**. if not self._is_blocked_on_rx_chan(): - cs.cancel() + cs.cancel( + reason=ContextCancelReason( + message=message, + error=error, + cid=self.cid, + actor_uid=self._actor.aid.uid, + peer_uid=self.chan.aid.uid, + scope_token=self._scope_token, + origin=origin, + ) + ) else: # rx_stats = self._rx_chan.statistics() message: str = 'NOT cancelling `Context._scope` !\n\n' @@ -1077,7 +1213,7 @@ class Context: ''' side: str = self.side - self._cancel_called = True + self._status._cancel_called = True # ^ XXX for debug via the `@.setter` # self.cancel_called = True @@ -1196,11 +1332,11 @@ class Context: our peer (task); if set immediately raise. This is a convenience wrapper for - `._maybe_raise_remote_err(self._remote_error)`. + `._maybe_raise_remote_err(self._status._remote_error)`. ''' __tracebackhide__: bool = hide_tb - if re := self._remote_error: + if re := self._status._remote_error: return self._maybe_raise_remote_err( re, hide_tb=hide_tb, @@ -1304,13 +1440,13 @@ class Context: # NOTE: we set the local scope error to any "self # cancellation" error-response thus "absorbing" # the error silently B) - if self._local_error is None: - self._local_error = remote_error + if self._status._local_error is None: + self._status._local_error = remote_error else: log.warning( 'Local error already set for ctx?\n' - f'{self._local_error}\n\n' + f'{self._status._local_error}\n\n' f'{self}' ) @@ -1373,8 +1509,8 @@ class Context: 'Invalid usage of `Context.wait_for_result()`!\n' 'Not valid on child-side IPC ctx!\n' ) - if self._final_result_is_set(): - return self._result + if self._status._final_result_is_set(): + return self._status._result assert self._rx_chan raise_overrun: bool = not self._allow_overruns @@ -1408,12 +1544,12 @@ class Context: ) # ?XXX, should already be set in `._deliver_msg()` right? - if self._outcome_msg is not Unresolved: + if self._status._outcome_msg is not Unresolved: # from .devx import debug # await debug.pause() - assert self._outcome_msg is outcome_msg + assert self._status._outcome_msg is outcome_msg else: - self._outcome_msg = outcome_msg + self._status._outcome_msg = outcome_msg if drained_msgs: drained_status += ( @@ -1435,7 +1571,7 @@ class Context: # should we raise overruns, bc ow we're # raising something we know might happen # during cancellation ;) - (not self._cancel_called) + (not self._status._cancel_called) ), hide_tb=hide_tb, ) @@ -1469,8 +1605,10 @@ class Context: Remote errors take precedence over local ones. ''' - le: BaseException|None = self._local_error - re: RemoteActorError|ContextCancelled|None = self._remote_error + le: BaseException|None = self._status._local_error + re: RemoteActorError|ContextCancelled|None = ( + self._status._remote_error + ) match (le, re): # NOTE: remote errors always get precedence since even @@ -1501,7 +1639,7 @@ class Context: if error: return error - if cancmsg := self._cancel_msg: + if cancmsg := self._status._cancel_msg: # NOTE: means we're prolly in the process of # processing the cancellation caused by # this msg (eg. logging from `Actor._cancel_task()` @@ -1510,16 +1648,15 @@ class Context: # without it eventually resulting in this property # delivering a value! log.debug( - '`Context._cancel_msg` is set but has not yet resolved to `.maybe_error`?\n\n' + '`ContextStatus._cancel_msg` is set but has not yet ' + 'resolved to `Context.maybe_error`?\n' + '\n' f'{cancmsg}\n' ) - # assert not self._cancel_msg + # assert not self._status._cancel_msg return None - def _final_result_is_set(self) -> bool: - return self._result is not Unresolved - # def get_result_nowait(self) -> Any|None: # def get_outcome_nowait(self) -> Any|None: # def recv_result_nowait(self) -> Any|None: @@ -1554,12 +1691,15 @@ class Context: return ( self.maybe_error or - self._result + self._status._result ) @property def has_outcome(self) -> bool: - return bool(self.maybe_error) or self._final_result_is_set() + return ( + bool(self.maybe_error) + or self._status._final_result_is_set() + ) def repr_outcome( self, @@ -1608,9 +1748,9 @@ class Context: ) return ( - str(self._result) + str(self._status._result) or - repr(self._result) + repr(self._status._result) ) @property @@ -1622,7 +1762,7 @@ class Context: ''' merr: Exception|None = self.maybe_error - outcome: Unresolved|Exception|Any = self.outcome + outcome: type[Unresolved]|Exception|Any = self.outcome status: str|None = None match ( outcome, @@ -1637,7 +1777,7 @@ class Context: status = 'self-cancelled' elif ( self.canceller - and not self._cancel_called + and not self._status._cancel_called ): status = 'peer-cancelled' @@ -1658,7 +1798,7 @@ class Context: case ( _, # any non-unresolved value None, - ) if self._final_result_is_set(): + ) if self._status._final_result_is_set(): status = 'result-returned' # normal operation but still in a pre-`Return`-result @@ -1673,7 +1813,7 @@ class Context: else: status = 'streaming' - elif self._started_called: + elif self._status._started_called: status = 'started' else: @@ -1716,7 +1856,7 @@ class Context: f'Caller side context {self} can not call started!' ) - elif self._started_called: + elif self._status._started_called: raise RuntimeError( f'called `.started()` twice on context with {self.chan.aid.uid}' ) @@ -1764,9 +1904,9 @@ class Context: await self.chan.send(started_msg) # set msg-related internal runtime-state - self._started_called: bool = True - self._started_msg: Started = started_msg - self._started_pld = value + self._status._started_called: bool = True + self._status._started_msg: Started = started_msg + self._status._started_pld = value async def _drain_overflows( self, @@ -1831,7 +1971,7 @@ class Context: If `._allow_overruns == True` (maybe) append the msg to an "overflow queue" and start a "drainer task" (inside the - `._scope_nursery: trio.Nursery`) which ensures that such + `._scope_tn: trio.Nursery`) which ensures that such messages are queued up and eventually sent if possible. XXX RULES XXX @@ -1883,7 +2023,7 @@ class Context: f'{flow_body}' f'{pformat(re)}\n' ) - self._cancel_msg: dict = msg + self._status._cancel_msg: dict = msg # XXX NOTE: this will not raise an error, merely set # `._remote_error` and maybe cancel any task currently @@ -1946,7 +2086,7 @@ class Context: stream._stop_msg = msg case Return(): - if self._outcome_msg is Unresolved: + if self._status._outcome_msg is Unresolved: log.warning( f'Setting final outcome msg AFTER ' f'`._send_chan.send_nowait()`??\n' @@ -1956,7 +2096,7 @@ class Context: # 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 + self._status._outcome_msg = msg return True @@ -2009,7 +2149,7 @@ class Context: f' |_cid: {cid}\n' f' |_task: {self._task}\n' ) - if not self._stream_opened: + if not self._status._stream_opened: txt += ( f'\n' f'*** No stream open on `{local_aid.name}` side! ***\n' @@ -2030,7 +2170,7 @@ class Context: not self._in_overrun ): self._overflow_q.append(msg) - tn: trio.Nursery = self._scope_nursery + tn: trio.Nursery = self._scope_tn assert not tn.child_tasks try: tn.start_soon( @@ -2222,7 +2362,7 @@ async def open_context_from_portal( # NOTE: this in an implicit runtime nursery used to, # - start overrun queuing tasks when as well as # for cancellation of the scope opened by the user. - ctx._scope_nursery: trio.Nursery = tn + ctx._scope_tn: trio.Nursery = tn ctx._scope: trio.CancelScope = tn.cancel_scope # XXX NOTE since `._scope` is NOT set BEFORE we retreive the @@ -2272,9 +2412,9 @@ async def open_context_from_portal( 'Invalid cancellation during IPC ctx sync phase?\n' ) from taskc - ctx._started_called: bool = True - ctx._started_msg: bool = started_msg - ctx._started_pld: bool = first + ctx._status._started_called: bool = True + ctx._status._started_msg: bool = started_msg + ctx._status._started_pld: bool = first # deliver context ref and `.started()` msg payload value # in `__aenter__` tuple. @@ -2334,7 +2474,7 @@ async def open_context_from_portal( # of a `Context`. In both cases any `ContextCancelled` # raised in this scope-block came from a transport msg # relayed from some remote-actor-task which our runtime set - # as to `Context._remote_error` + # as to `ContextStatus._remote_error` # # the CASES: # @@ -2366,7 +2506,7 @@ async def open_context_from_portal( # `Nursery.cancel_scope.cancel()`) except ContextCancelled as ctxc: scope_err = ctxc - ctx._local_error: BaseException = scope_err + ctx._status._local_error: BaseException = scope_err ctxc_from_child = ctxc # XXX TODO XXX: FIX THIS debug_mode BUGGGG!!! @@ -2384,9 +2524,9 @@ async def open_context_from_portal( # `.cancel()`, we don't raise and the exit block should # finish silently. if ( - ctx._cancel_called + ctx._status._cancel_called and - ctxc is ctx._remote_error + ctxc is ctx._status._remote_error and ctxc.canceller == portal.actor.aid.uid ): @@ -2411,7 +2551,7 @@ async def open_context_from_portal( # CASE 3: standard local error in this parent/yieldee Exception, - # CASES 1 & 2: can manifest as a `ctx._scope_nursery` + # CASES 1 & 2: can manifest as a `ctx._scope_tn` # exception-group of, # # 1.-`trio.Cancelled`s, since @@ -2439,7 +2579,7 @@ async def open_context_from_portal( ) as rent_err: scope_err = rent_err - ctx._local_error: BaseException = scope_err + ctx._status._local_error: BaseException = scope_err # XXX: ALWAYS request the context to CANCEL ON any ERROR. # NOTE: `Context.cancel()` is conversely NEVER CALLED in @@ -2509,7 +2649,7 @@ async def open_context_from_portal( # `Context.wait_for_result()` will ALWAYS raise # a `ContextCancelled` (via an embedded call to # `Context._maybe_raise_remote_err()`) IFF - # a `Context._remote_error` was set by the runtime + # a `ContextStatus._remote_error` was set by the runtime # via a call to # `Context._maybe_cancel_and_set_remote_error()`. # As per `Context._deliver_msg()`, that error IS @@ -2526,7 +2666,7 @@ async def open_context_from_portal( # this task didn't know until final teardown # / value collection. scope_err = berr - ctx._local_error: BaseException = scope_err + ctx._status._local_error: BaseException = scope_err raise # yes this worx! @@ -2624,17 +2764,17 @@ async def open_context_from_portal( # generally speaking mask runtime-machinery related # multi-`trio.Cancelled`s. As such, any `scope_error` # which was the underlying cause of this context's exit - # should be stored as the `Context._local_error` and + # should be stored as the `ContextStatus._local_error` and # used in determining `Context.cancelled_caught: bool`. if scope_err is not None: # sanity, tho can remove? - assert ctx._local_error is scope_err - # ctx._local_error: BaseException = scope_err + assert ctx._status._local_error is scope_err + # ctx._status._local_error: BaseException = scope_err # etype: Type[BaseException] = type(scope_err) # CASE 2 if ( - ctx._cancel_called + ctx._status._cancel_called and ctx.cancel_acked ): log.cancel( @@ -2720,7 +2860,6 @@ def mk_context( _caller_info=caller_info, **kwargs, ) - ctx._result = Unresolved return ctx diff --git a/tractor/_exceptions.py b/tractor/_exceptions.py index 52cec604..9434676d 100644 --- a/tractor/_exceptions.py +++ b/tractor/_exceptions.py @@ -1406,7 +1406,12 @@ def _raise_from_unexpected_msg( msg, ctx.chan, ) - ctx._maybe_cancel_and_set_remote_error(exc) + ctx._maybe_cancel_and_set_remote_error( + exc, + # PldRx emulates a remote Error for local validation + # failures; src_uid alone cannot identify the observer. + local_protocol=isinstance(src_err, ValidationError), + ) raise exc from src_err # `MsgStream` termination msg. diff --git a/tractor/_streaming.py b/tractor/_streaming.py index 93c7d12f..6c122eb8 100644 --- a/tractor/_streaming.py +++ b/tractor/_streaming.py @@ -190,8 +190,8 @@ class MsgStream(trio.abc.Channel): # from the `.aclose()` caller? # # breakpoint() # to debug this RACE CASE! - ctx._result = pld - ctx._outcome_msg = msg + ctx._status._result = pld + ctx._status._outcome_msg = msg return pld @@ -709,7 +709,7 @@ async def open_stream_from_ctx( # since it likely means the surrounding lexical-scope has # errored, been `trio.Cancelled` or at the least # `Context.cancel()` was called by some task. - if ctx._cancel_called: + if ctx._status._cancel_called: # XXX NOTE: ALWAYS RAISE any remote error here even if # it's an expected `ContextCancelled` due to a local @@ -728,8 +728,8 @@ async def open_stream_from_ctx( # because we want to raise a ctxc on any task entering this `.open_stream()` # AFTER cancellation was already been requested, # we DO NOT want to absorb any ctxc ACK silently! - # if ctx._remote_error: - # raise ctx._remote_error + # if ctx._status._remote_error: + # raise ctx._status._remote_error # XXX NOTE: if no `ContextCancelled` has been responded # back from the other side (yet), we raise a different @@ -744,7 +744,7 @@ async def open_stream_from_ctx( if ( not ctx._portal - and not ctx._started_called + and not ctx._status._started_called ): raise RuntimeError( 'Context.started()` must be called before opening a stream' @@ -803,7 +803,7 @@ async def open_stream_from_ctx( ctx._portal._streams.add(stream) try: - ctx._stream_opened: bool = True + ctx._status._stream_opened: bool = True ctx._stream = stream # XXX: do we need this? diff --git a/tractor/devx/debug/_trace.py b/tractor/devx/debug/_trace.py index 5bac9d6c..dc31fa92 100644 --- a/tractor/devx/debug/_trace.py +++ b/tractor/devx/debug/_trace.py @@ -696,7 +696,7 @@ async def pause( api_frame: FrameType = api_frame or inspect.currentframe() # XXX TODO: this was causing cs-stack corruption in trio due to - # usage within the `Context._scope_nursery` (which won't work + # usage within the `Context._scope_tn` (which won't work # based on scoping of it versus call to `_maybe_enter_debugger()` # from `._rpc._invoke()`) # with trio.CancelScope( diff --git a/tractor/devx/debug/_tty_lock.py b/tractor/devx/debug/_tty_lock.py index 0b63798d..14d551fd 100644 --- a/tractor/devx/debug/_tty_lock.py +++ b/tractor/devx/debug/_tty_lock.py @@ -1024,7 +1024,7 @@ async def request_root_stdio_lock( # `bdb.BdbQuit` directly, OW you get a trio cs stack # corruption! # Further, the since this task is spawned inside the - # `Context._scope_nursery: trio.Nursery`, once an RPC + # `Context._scope_tn: trio.Nursery`, once an RPC # task errors that cs is cancel_called and so if we want # to debug the TPC task that failed we need to shield # against that expected `.cancel()` call and instead diff --git a/tractor/msg/_ops.py b/tractor/msg/_ops.py index 1b342ca6..86e5e649 100644 --- a/tractor/msg/_ops.py +++ b/tractor/msg/_ops.py @@ -593,7 +593,7 @@ async def drain_to_final_msg( while not ( ctx.maybe_error and - not ctx._final_result_is_set() + not ctx._status._final_result_is_set() ): try: # receive all msgs, scanning for either a final result @@ -678,7 +678,7 @@ async def drain_to_final_msg( f'\n' f'{pretty_struct.pformat(msg)}' ) - ctx._result: Any = pld + ctx._status._result: Any = pld result_msg = msg break @@ -697,7 +697,7 @@ async def drain_to_final_msg( and (reason := 'ctx cancelled other side') ) - or (ctx._cancel_called + or (ctx._status._cancel_called and (reason := 'ctx called `.cancel()`') ) @@ -768,9 +768,9 @@ async def drain_to_final_msg( # if raises: # log.error('some msg about raising..') # - re: Exception|None = ctx._remote_error + re: Exception|None = ctx._status._remote_error if re: - assert msg is ctx._cancel_msg + assert msg is ctx._status._cancel_msg # NOTE: this solved a super duper edge case XD # this was THE super duper edge case of: # - local task opens a remote task, diff --git a/tractor/runtime/_rpc.py b/tractor/runtime/_rpc.py index 70c96955..d030a570 100644 --- a/tractor/runtime/_rpc.py +++ b/tractor/runtime/_rpc.py @@ -362,7 +362,7 @@ async def _errors_relayed_via_ipc( not isinstance(err, ContextCancelled) or ( isinstance(err, ContextCancelled) - and ctx._cancel_called + and ctx._status._cancel_called # if the root blocks the debugger lock request from a child # we will get a remote-cancelled condition. @@ -720,7 +720,7 @@ async def _invoke( unmask_from=(Cancelled,), ) as _mbme, # maybe boxed masked exc ): - ctx._scope_nursery = tn + ctx._scope_tn = tn rpc_ctx_cs = ctx._scope = tn.cancel_scope _register_rpc_task( actor, @@ -738,7 +738,7 @@ async def _invoke( pld=res, ) # set and shuttle final result to "parent"-side task. - ctx._result = res + ctx._status._result = res log.runtime( f'Sending result msg and exiting {ctx.side!r}\n' f'\n' @@ -791,7 +791,7 @@ async def _invoke( # before raising any context cancelled case # so that real remote errors don't get masked as # ``ContextCancelled``s. - if re := ctx._remote_error: + if re := ctx._status._remote_error: ctx._maybe_raise_remote_err(re) if rpc_ctx_cs.cancel_called: @@ -803,7 +803,7 @@ async def _invoke( # ``Context.cancel()`` directly, we're going to # need to change this logic branch since it # will always enter.. - if ctx._cancel_called: + if ctx._status._cancel_called: # TODO: test for this!!!!! canceller: tuple = our_uid explain += 'itself ' @@ -879,7 +879,7 @@ async def _invoke( # always set this (child) side's exception as the # local error on the context - ctx._local_error: BaseException = scope_err + ctx._status._local_error: BaseException = scope_err # ^-TODO-^ question, # does this matter other then for # consistentcy/testing? @@ -1392,9 +1392,9 @@ async def process_messages( 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() + and ctx._status._remote_error is None + and ctx._status._outcome_msg is Unresolved + and not ctx._status._final_result_is_set() ): ctx._maybe_cancel_and_set_remote_error(tc) diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index a9f1c8f7..bf7dee3c 100644 --- a/tractor/runtime/_runtime.py +++ b/tractor/runtime/_runtime.py @@ -1428,20 +1428,20 @@ class Actor: # f' >> {ctx._nsf}()\n' ) if ( - ctx._canceller is None + ctx._status._canceller is None and requesting_aid ): - ctx._canceller: tuple = requesting_aid.uid + ctx._status._canceller: tuple = requesting_aid.uid # TODO: pack the RPC `{'cmd': }` msg into a ctxc and # then raise and pack it here? if ( ipc_msg - and ctx._cancel_msg is None + and ctx._status._cancel_msg is None ): # assign RPC msg directly from the loop which usually # the case with `ctx.cancel()` on the other side. - ctx._cancel_msg = ipc_msg + ctx._status._cancel_msg = ipc_msg # don't allow cancelling this function mid-execution # (is this necessary?)