Clean failed remote-task startup state
`Actor.start_remote_task()` registers its caller context before sending `Start`, but only cancellation cleaned that state. Encoding, ack timeout, malformed ack and remote authorization errors leaked it. Protect the complete send, acknowledgement and validation phase. Track successful publication, make a remote cancellation attempt only when protocol-safe and always release the local context while preserving the original startup error. Cover both pre-publication serialization failure and a remote `ModuleNotExposed` rejection without damaging a reused portal. Caught-during: review remediation Found-via: `/run-tests` startup-failure regressions Review: PR #481 (opencode) https://github.com/goodboy/tractor/pull/481#pullrequestreview-4956692120 Prompt-IO: ai/prompt-io/opencode/20260818T193004Z_bf06b4f8_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))drop_ria_nursery
parent
96e4934573
commit
51a2b7a4f8
|
|
@ -0,0 +1,36 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-5.6-sol
|
||||||
|
service: opencode
|
||||||
|
session: ses_3e4c90d3eafeqHEtRYSIHgHhpA
|
||||||
|
timestamp: 2026-08-18T19:30:04Z
|
||||||
|
git_ref: bf06b4f8
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260818T193004Z_bf06b4f8_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Release caller-side context state for every remote-task startup failure,
|
||||||
|
not only local cancellation. Preserve the remote error, avoid unsafe
|
||||||
|
follow-up sends and prove pre-publication serialization failures leave
|
||||||
|
a reused portal healthy.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Extend remote-task startup cleanup across send, acknowledgement and
|
||||||
|
validation errors. Track completed publication, perform only safe
|
||||||
|
best-effort cancellation and deterministically remove local state.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `tractor/runtime/_runtime.py` - clean every startup failure path.
|
||||||
|
- `tests/test_context_stream_semantics.py` - cover authorization and
|
||||||
|
serialization failures before context entry.
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
The human accepted the discovered edge-case fixes but required general
|
||||||
|
startup cleanup to land separately from cancellation cleanup, transport
|
||||||
|
integrity and the public API. This boundary preserves that behavioral
|
||||||
|
distinction and its dedicated commit-message rationale.
|
||||||
|
|
@ -0,0 +1,19 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-5.6-sol
|
||||||
|
service: opencode
|
||||||
|
timestamp: 2026-08-18T19:30:04Z
|
||||||
|
git_ref: bf06b4f8
|
||||||
|
diff_cmd: git diff HEAD~1..HEAD
|
||||||
|
---
|
||||||
|
|
||||||
|
`Actor.start_remote_task()` inserts a context before sending `Start`,
|
||||||
|
but startup errors other than cancellation escape without removing or
|
||||||
|
closing that caller state. Serialization errors, acknowledgement
|
||||||
|
timeouts, malformed acknowledgements and remote authorization errors
|
||||||
|
can therefore leak context-registry entries.
|
||||||
|
|
||||||
|
> `git diff HEAD~1..HEAD -- tractor/runtime/_runtime.py tests/test_context_stream_semantics.py`
|
||||||
|
|
||||||
|
Cover the complete send, acknowledgement and validation phase with
|
||||||
|
exceptional cleanup. Attempt remote cancellation only when publication
|
||||||
|
is known complete or protocol-safe, and always release local state.
|
||||||
|
|
@ -286,6 +286,82 @@ async def test_cancel_during_context_startup(
|
||||||
await portal.cancel_actor()
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_start_serialization_error_cleans_context(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Deallocate caller state when `Start` can not be serialized.
|
||||||
|
|
||||||
|
`Actor.start_remote_task()` registers its caller-side `Context`
|
||||||
|
before encoding the request. An unsupported argument used to raise
|
||||||
|
`MsgTypeError` before publication while leaking that registry
|
||||||
|
entry. Comparing the context registry around the failed start
|
||||||
|
proves cleanup, and a following valid context proves no bytes
|
||||||
|
reached or damaged the reused portal's transport.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
actor = tractor.current_actor()
|
||||||
|
portal: tractor.Portal = await an.start_actor(
|
||||||
|
'serialization_error_worker',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
contexts_before = _non_registration_contexts(actor)
|
||||||
|
with pytest.raises(tractor.MsgTypeError):
|
||||||
|
async with portal.open_context(
|
||||||
|
simple_setup_teardown,
|
||||||
|
data=object(),
|
||||||
|
):
|
||||||
|
raise AssertionError('invalid `Start` was accepted')
|
||||||
|
|
||||||
|
assert _non_registration_contexts(actor) == contexts_before
|
||||||
|
async with portal.open_context(
|
||||||
|
simple_setup_teardown,
|
||||||
|
data=1,
|
||||||
|
) as (ctx, started):
|
||||||
|
assert started == 2
|
||||||
|
assert await ctx.wait_for_result() == 'yo'
|
||||||
|
|
||||||
|
assert _non_registration_contexts(actor) == contexts_before
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_start_module_error_cleans_context(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Deallocate caller state after a remote startup rejection.
|
||||||
|
|
||||||
|
A target actor without this test module rejects the requested
|
||||||
|
context before sending `StartAck`. That remote
|
||||||
|
`ModuleNotExposed` used to escape startup validation while leaving
|
||||||
|
the caller context registered. The boxed error and before/after
|
||||||
|
registry comparison prove the remote failure remains visible and
|
||||||
|
local startup state is released.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
actor = tractor.current_actor()
|
||||||
|
portal: tractor.Portal = await an.start_actor(
|
||||||
|
'module_error_worker',
|
||||||
|
)
|
||||||
|
contexts_before = _non_registration_contexts(actor)
|
||||||
|
with pytest.raises(tractor.RemoteActorError) as excinfo:
|
||||||
|
async with portal.open_context(
|
||||||
|
simple_setup_teardown,
|
||||||
|
data=1,
|
||||||
|
):
|
||||||
|
raise AssertionError('unexposed context was started')
|
||||||
|
|
||||||
|
assert excinfo.value.boxed_type is tractor.ModuleNotExposed
|
||||||
|
assert _non_registration_contexts(actor) == contexts_before
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
'error_parent',
|
'error_parent',
|
||||||
[False, ValueError, KeyboardInterrupt],
|
[False, ValueError, KeyboardInterrupt],
|
||||||
|
|
|
||||||
|
|
@ -836,8 +836,10 @@ class Actor:
|
||||||
|
|
||||||
f'{pretty_struct.pformat(msg)}'
|
f'{pretty_struct.pformat(msg)}'
|
||||||
)
|
)
|
||||||
|
start_published: bool = False
|
||||||
try:
|
try:
|
||||||
await chan.send(msg)
|
await chan.send(msg)
|
||||||
|
start_published = True
|
||||||
|
|
||||||
# NOTE wait on first `StartAck` response msg and validate;
|
# NOTE wait on first `StartAck` response msg and validate;
|
||||||
# this should be immediate and does not (yet) wait for the
|
# this should be immediate and does not (yet) wait for the
|
||||||
|
|
@ -845,7 +847,22 @@ class Actor:
|
||||||
with trio.fail_after(ack_timeout):
|
with trio.fail_after(ack_timeout):
|
||||||
first_msg: msgtypes.StartAck = await ctx._rx_chan.receive()
|
first_msg: msgtypes.StartAck = await ctx._rx_chan.receive()
|
||||||
|
|
||||||
except trio.Cancelled:
|
try:
|
||||||
|
functype: str = first_msg.functype
|
||||||
|
except AttributeError:
|
||||||
|
raise unpack_error(first_msg, chan)
|
||||||
|
|
||||||
|
if functype not in (
|
||||||
|
'asyncfunc',
|
||||||
|
'asyncgen',
|
||||||
|
'context',
|
||||||
|
):
|
||||||
|
raise ValueError(
|
||||||
|
f'Invalid `StartAck.functype: str = '
|
||||||
|
f'{first_msg!r}` ??'
|
||||||
|
)
|
||||||
|
|
||||||
|
except BaseException as startup_err:
|
||||||
with trio.CancelScope(shield=True):
|
with trio.CancelScope(shield=True):
|
||||||
# `MsgpackTransport.send()` closes its stream when
|
# `MsgpackTransport.send()` closes its stream when
|
||||||
# cancellation interrupts the length-prefixed write
|
# cancellation interrupts the length-prefixed write
|
||||||
|
|
@ -856,7 +873,14 @@ class Actor:
|
||||||
if (
|
if (
|
||||||
cancel_on_startup
|
cancel_on_startup
|
||||||
and
|
and
|
||||||
chan.connected()
|
(
|
||||||
|
start_published
|
||||||
|
or (
|
||||||
|
isinstance(startup_err, trio.Cancelled)
|
||||||
|
and
|
||||||
|
chan.connected()
|
||||||
|
)
|
||||||
|
)
|
||||||
):
|
):
|
||||||
try:
|
try:
|
||||||
await ctx.cancel()
|
await ctx.cancel()
|
||||||
|
|
@ -872,19 +896,6 @@ class Actor:
|
||||||
await ctx._rx_chan.aclose()
|
await ctx._rx_chan.aclose()
|
||||||
|
|
||||||
raise
|
raise
|
||||||
try:
|
|
||||||
functype: str = first_msg.functype
|
|
||||||
except AttributeError:
|
|
||||||
raise unpack_error(first_msg, chan)
|
|
||||||
|
|
||||||
if functype not in (
|
|
||||||
'asyncfunc',
|
|
||||||
'asyncgen',
|
|
||||||
'context',
|
|
||||||
):
|
|
||||||
raise ValueError(
|
|
||||||
f'Invalid `StartAck.functype: str = {first_msg!r}` ??'
|
|
||||||
)
|
|
||||||
|
|
||||||
ctx._remote_func_type = functype
|
ctx._remote_func_type = functype
|
||||||
return ctx
|
return ctx
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue