forked from goodboy/tractor
1
0
Fork 0

Quieter `Stop` handling on ctx result capture

In the `drain_to_final_msg()` impl, since a stream terminating
gracefully requires this msg, there's really no reason to `log.cancel()`
about it; go `.runtime()` level instead since we're trying de-noise
under "normal operation".

Also,
- passthrough `hide_tb` to taskc-handler's `ctx.maybe_raise()` call.
- raise `MessagingError` for the `MsgType` unmatched `case _:`.
- detail the doc string motivation a little more.
aio_abandons
Tyler Goodlet 2024-07-03 17:01:37 -04:00
parent 9be821a5cf
commit b56352b0e4
1 changed files with 36 additions and 19 deletions

View File

@ -374,7 +374,7 @@ class PldRx(Struct):
case _: case _:
src_err = InternalError( src_err = InternalError(
'Unknown IPC msg ??\n\n' 'Invalid IPC msg ??\n\n'
f'{msg}\n' f'{msg}\n'
) )
@ -499,7 +499,7 @@ async def maybe_limit_plds(
yield None yield None
return return
# sanity on scoping # sanity check on IPC scoping
curr_ctx: Context = current_ipc_ctx() curr_ctx: Context = current_ipc_ctx()
assert ctx is curr_ctx assert ctx is curr_ctx
@ -510,6 +510,8 @@ async def maybe_limit_plds(
) as msgdec: ) as msgdec:
yield msgdec yield msgdec
# when the applied spec is unwound/removed, the same IPC-ctx
# should still be in scope.
curr_ctx: Context = current_ipc_ctx() curr_ctx: Context = current_ipc_ctx()
assert ctx is curr_ctx assert ctx is curr_ctx
@ -525,16 +527,26 @@ async def drain_to_final_msg(
list[MsgType] list[MsgType]
]: ]:
''' '''
Drain IPC msgs delivered to the underlying IPC primitive's Drain IPC msgs delivered to the underlying IPC context's
rx-mem-chan (eg. `Context._rx_chan`) from the runtime in rx-mem-chan (i.e. from `Context._rx_chan`) in search for a final
search for a final result or error. `Return` or `Error` msg.
The motivation here is to ideally capture errors during ctxc Deliver the `Return` + preceding drained msgs (`list[MsgType]`)
conditions where a canc-request/or local error is sent but the as a pair unless an `Error` is found, in which unpack and raise
local task also excepts and enters the it.
`Portal.open_context().__aexit__()` block wherein we prefer to
capture and raise any remote error or ctxc-ack as part of the The motivation here is to always capture any remote error relayed
`ctx.result()` cleanup and teardown sequence. by the remote peer task during a ctxc condition.
For eg. a ctxc-request may be sent to the peer as part of the
local task's (request for) cancellation but then that same task
**also errors** before executing the teardown in the
`Portal.open_context().__aexit__()` block. In such error-on-exit
cases we want to always capture and raise any delivered remote
error (like an expected ctxc-ACK) as part of the final
`ctx.wait_for_result()` teardown sequence such that the
`Context.outcome` related state always reflect what transpired
even after ctx closure and the `.open_context()` block exit.
''' '''
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
@ -572,7 +584,6 @@ async def drain_to_final_msg(
# |_from tractor.devx._debug import pause # |_from tractor.devx._debug import pause
# await pause() # await pause()
# NOTE: we get here if the far end was # NOTE: we get here if the far end was
# `ContextCancelled` in 2 cases: # `ContextCancelled` in 2 cases:
# 1. we requested the cancellation and thus # 1. we requested the cancellation and thus
@ -580,13 +591,13 @@ async def drain_to_final_msg(
# 2. WE DID NOT REQUEST that cancel and thus # 2. WE DID NOT REQUEST that cancel and thus
# SHOULD RAISE HERE! # SHOULD RAISE HERE!
except trio.Cancelled as taskc: except trio.Cancelled as taskc:
# CASE 2: mask the local cancelled-error(s) # CASE 2: mask the local cancelled-error(s)
# only when we are sure the remote error is # only when we are sure the remote error is
# the source cause of this local task's # the source cause of this local task's
# cancellation. # cancellation.
ctx.maybe_raise( ctx.maybe_raise(
# TODO: when use this/ hide_tb=hide_tb,
# TODO: when use this?
# from_src_exc=taskc, # from_src_exc=taskc,
) )
@ -659,7 +670,7 @@ async def drain_to_final_msg(
# Stop() # Stop()
case Stop(): case Stop():
pre_result_drained.append(msg) pre_result_drained.append(msg)
log.cancel( log.runtime( # normal/expected shutdown transaction
'Remote stream terminated due to "stop" msg:\n\n' 'Remote stream terminated due to "stop" msg:\n\n'
f'{pretty_struct.pformat(msg)}\n' f'{pretty_struct.pformat(msg)}\n'
) )
@ -719,13 +730,19 @@ async def drain_to_final_msg(
pre_result_drained.append(msg) pre_result_drained.append(msg)
# It's definitely an internal error if any other # It's definitely an internal error if any other
# msg type without a`'cid'` field arrives here! # msg type without a`'cid'` field arrives here!
report: str = (
f'Invalid or unknown msg type {type(msg)!r}!?\n'
)
if not msg.cid: if not msg.cid:
raise InternalError( report += (
'Unexpected cid-missing msg?\n\n' '\nWhich also has no `.cid` field?\n'
f'{msg}\n'
) )
raise RuntimeError('Unknown msg type: {msg}') raise MessagingError(
report
+
f'\n{msg}\n'
)
else: else:
log.cancel( log.cancel(