Compare commits

...

11 Commits

Author SHA1 Message Date
Gud Boi 203a0f7e1f Add `.trionics.start_or_cancel()` test suite
Resolve #474 with a new `tests/trionics/test_taskc.py` (9
tests) covering the `trio.Nursery.start()` wrapper landed in
PR #464, incl. the `modden.runtime.progman.open_wks()` use
case dug out as a minimal repro.

Deats,
- the lossy `RuntimeError('child exited without calling
  task_status.started()')` only fires when the child absorbs
  its ambient cancel pre-`.started()` (graceful-teardown
  pattern); a well-behaved child surfaces `Cancelled` direct
  from `.start()` on `trio` 0.29 - verified empirically 1st.
- `test_sibling_err_not_masked_by_startup_rte`: the `modden`
  case; ONLY the root-cause sibling `ValueError` escapes the
  nursery with the wrapper vs. bare-`.start()`'s lossy
  riding-along startup-RTE noise.
- `test_pure_oob_cancel_not_morphed_to_rte`: plain ancestor
  `cs.cancel()` exits clean vs. bare's eg-wrapped RTE.
- `test_genuine_startup_rte_still_raised`: sans cancellation
  the protocol-bug RTE re-raises same as bare.
- `test_childs_own_rte_never_demoted_to_cancel`: exact-msg +
  `isinstance`-guard regression cover; a child's own
  `RuntimeError('never got started!')`/`RuntimeError(1234)`
  never demotes to a `Cancelled`.
- `test_started_value_and_args_passthru`: positional args,
  `name=` and `.started()`-value forwarding.

Also,
- `use_start_or_cancel=False` params pin upstream `trio`'s
  current lossy behaviour as wart-documentation: a break on
  a `trio` upgrade likely means new upstream porcelain and
  the wrapper deserves a re-audit.
- verified 0 flakes over 50 hammer runs + 2 impl mutations
  each caught by exactly the targeted tests.

Prompt-IO: ai/prompt-io/claude/20260702T161624Z_65bf9df5_prompt_io.md
(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-08-12 14:15:31 -04:00
Bd 92c737ad83
Merge pull request #458 from mahmoudhas/fix/guard-hot-path-log-rendering
Guard hot-path log calls to avoid payload rendering when disabled
2026-08-12 14:08:47 -04:00
Gud Boi 935c8cf656 Guard receive-path transport rendering
Skip raw packet, decoded message, peer, and channel formatting when
transport logging is disabled. Keep message processing and wire
reads outside the guards so logging controls never affect IPC flow.

Also narrow the remaining pretty-struct TODO to require a
non-raising formatter with native-repr fallback.

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-12 13:48:40 -04:00
Gud Boi 84ec895150 Drop resolved channel log-guard TODO
Remove the stale `Channel.from_addr()` design note now that its
`at_least_level()` guard avoids inactive pretty-rendering work.

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-12 12:51:43 -04:00
Gud Boi 67280c2898 Honor log disable controls in hot-path guards
Use `Logger.isEnabledFor()` in `at_least_level()` so logger-local
and global disable controls short-circuit payload rendering.

Add `Channel.send()` coverage for effective-level, per-logger, and
global suppression while ensuring transport remains unchanged.

Caught-during: review remediation
Found-via: `/run-tests` test_log_guard_skips_payload_formatting

Review: PR #458 (goodboy)
https://github.com/goodboy/tractor/pull/458#issuecomment-5258207470

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-11 21:56:27 -04:00
root 0e11ff7e9d Guard hot-path log calls to avoid payload rendering when disabled
Wrap `log.transport()` in `Channel.send()` and `log.runtime()` in
`PldRx.decode_pld()` with `log.at_least_level()` checks so that
expensive `pformat(payload)` / `repr(msg)` / `repr(pld)` calls are
skipped entirely when the respective log level is not active.

Previously the f-string arguments were eagerly evaluated before being
passed to the log method, even though `StackLevelAdapter.log()` would
then discard the message internally via its own `isEnabledFor()` check.
On high-frequency IPC paths this caused `pformat` to dominate CPU
usage (~60-70 %) as reported in #455.

This also restores the full diagnostic output (msg type, decoded
payload) that was temporarily commented out in 0373164 as a stopgap.

Resolves #455
2026-08-11 20:41:20 -04:00
mahmoud 148a098ca6 avoid format on the hot send path 2026-08-11 20:41:20 -04:00
Bd 83b3488455
Merge pull request #488 from goodboy/wkt/moc_teardown_completion
Wait for ctx exit in `maybe_open_context()`
2026-08-11 12:16:15 -04:00
Gud Boi daa661aba3 Clarify `maybe_open_context()` teardown notes
Drop the stale sentinel experiment and fix the cancellation-path
comment. Document that cached regular `__aexit__()` failures are
always re-raised at the final consumer boundary.

Review: PR #488 (goodboy)
https://github.com/goodboy/tractor/pull/488

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-11 11:38:43 -04:00
Gud Boi 55ec3dbf51 Drop unused `_Cache` teardown bindings
Remove the unused `value` assignment after cache eviction and skip
unpacking stale resource state before raising its invariant error.

Review: PR #488 (Copilot)
https://github.com/goodboy/tractor/pull/488#pullrequestreview-4850557500

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-07 22:24:27 -04:00
Gud Boi a753625fc6 Wait for ctx exit in `maybe_open_context()`
Block the final user on its cached resource's `__aexit__()` and
raise regular cleanup errors at that user's ctx boundary.

Deats,
- serialize user registration and teardown under each cache-key lock
- keep queued entrants on the same lock through resource replacement
- preserve `Cancelled`, `KeyboardInterrupt`, and `SystemExit` flow
- cover successful, failing, cancelled, and re-entry teardown paths

Prompt-IO: ai/prompt-io/opencode/20260804T030309Z_65bf9df5_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
2026-08-03 23:45:42 -04:00
14 changed files with 1107 additions and 115 deletions

View File

@ -0,0 +1,80 @@
---
model: claude-fable-5
service: claude
session: f4da6654-63f9-4641-b567-f200da46d697
timestamp: 2026-07-02T16:16:24Z
git_ref: 65bf9df5
scope: tests
substantive: true
raw_file: 20260702T161624Z_65bf9df5_prompt_io.raw.md
---
## Prompt
> attempt to resolve
> https://github.com/goodboy/tractor/issues/474
> do it with /open-wkt.
Issue #474: "We need test suite for
`.trionics.start_or_cancel()`" — dig out the use case
from `modden` as a unit test (and provide motivation
for upstream `trio` to possibly provide better
porcelain around this case). The wrapper landed via
PR #464 with no tests.
## Response summary
New `tests/trionics/test_taskc.py` (9 tests) covering
`start_or_cancel()`, authored in wkt
`start_or_cancel_tests_474`.
Behaviour was first probed empirically (trio 0.29):
the lossy `RuntimeError('child exited without calling
task_status.started()')` only fires when the child
exits pre-`.started()` WITHOUT propagating the ambient
`Cancelled` — i.e. when the child (or lib code it
calls) absorbs the cancel in a graceful-teardown
pattern; a well-behaved child surfaces `Cancelled`
straight out of `.start()`. The `modden`
`progman.open_wks()` use case was reconstructed from
`modden/runtime/progman.py` accordingly.
Tests (each `use_start_or_cancel` parametrization also
pins upstream trio's current lossy behaviour as
wart-documentation):
- `test_sibling_err_not_masked_by_startup_rte` — the
`modden` case: sibling error OOB-cancels the shared
nursery scope; with the wrapper ONLY the root-cause
`ValueError` escapes; bare `.start()` adds the lossy
RTE alongside.
- `test_pure_oob_cancel_not_morphed_to_rte` — plain
ancestor `cs.cancel()`: wrapper → clean exit; bare
→ eg-wrapped RTE.
- `test_genuine_startup_rte_still_raised` — no
cancellation → protocol-bug RTE re-raised same as
bare.
- `test_childs_own_rte_never_demoted_to_cancel` — a
child's own `RuntimeError('never got started!')` /
`RuntimeError(1234)` under ambient cancel is never
demoted to `Cancelled` (exact-msg-match + str-guard
regression cover).
- `test_started_value_and_args_passthru` — happy path:
positional args, `name=`, `.started()` value.
Verified: 9/9 pass; 0 flakes across 50 hammer runs;
two impl mutations (checkpoint removed; guard relaxed
to substring match) each caught by exactly the
targeted tests; `tests/trionics/` +
`tests/test_trioisms.py` subset green (23 passed,
5 xfailed); ruff clean; 69-col style.
## Files changed
- `tests/trionics/test_taskc.py` — new
`start_or_cancel()` unit-test suite (gh #474).
## Human edits
Pending review — session paused pre-commit per user
deadline; nothing committed as of this entry.

View File

@ -0,0 +1,107 @@
---
model: claude-fable-5
service: claude
timestamp: 2026-07-02T16:16:24Z
git_ref: 65bf9df5
diff_cmd: git diff main..wkt/start_or_cancel_tests_474
---
# Raw output — gh #474 `start_or_cancel()` test suite
## Generated test code
> `git diff main..wkt/start_or_cancel_tests_474 -- tests/trionics/test_taskc.py`
Prose summary of the generated module
(`tests/trionics/test_taskc.py`):
- module docstring framing the `trio.Nursery.start()`
startup-cancellation wart, the wrapper's repair, and
the intent that `use_start_or_cancel=False` params
double as upstream-trio wart-documentation (break on
a trio upgrade → upstream may have shipped porcelain,
re-audit the wrapper); cites gh #474 / PR #464 and
`modden`'s `progman.open_wks()` as the source use
case.
- shared children: `absorbs_cancel_pre_started()` (the
graceful-teardown cancel-absorber which triggers the
lossy RTE path) + `raise_val_err()` (fast-erroring
sibling).
- `test_sibling_err_not_masked_by_startup_rte`
(parametrized `use_start_or_cancel`): asserts eg
contains exactly one `ValueError` and, wrapper-case,
NO residual RTE (`eg.split(ValueError)` remainder is
`None`); bare-case, the residual RTE carries trio's
exact "child exited without calling" wording.
- `test_pure_oob_cancel_not_morphed_to_rte`
(parametrized): wrapper-case runs clean and asserts
`cs.cancelled_caught`; bare-case asserts the
eg-wrapped RTE.
- `test_genuine_startup_rte_still_raised`
(parametrized): no-cancel protocol bug → RTE with
trio's wording from both call forms.
- `test_childs_own_rte_never_demoted_to_cancel`
(parametrized `rte_arg` in `'never got started!'`,
`1234`): child cancels the ambient scope then raises
its own RTE synchronously (no checkpoint between →
deterministically under-cancellation at catch time);
asserts the RTE survives with `args[0]` intact.
- `test_started_value_and_args_passthru`: `.started()`
value, positional args and the `name=` kwarg (via
`trio.lowlevel.current_task().name`) all forward.
## Non-code output (verbatim highlights)
Behaviour probe (trio 0.29, scratchpad scripts) — the
decision basis for the test shapes:
```
== B-sibling-err use_soc=False
start raised: RuntimeError('child exited without
calling task_status.started()')
top-level: ExceptionGroup([ValueError('sibling blew
up!'), RuntimeError('child exited without calling
task_status.started()')])
== B-cs-cancel use_soc=False
top-level: ExceptionGroup([RuntimeError('child
exited without calling task_status.started()')])
== B-sibling-err use_soc=True
start raised: Cancelled()
top-level: ExceptionGroup([ValueError('sibling blew
up!')])
== B-cs-cancel use_soc=True
start raised: Cancelled()
top-level: clean return
== own-rte-under-cancel (both) -> RTE('never got
started!') propagates unchanged
```
Key finding: with a WELL-BEHAVED (non-absorbing) child
an OOB ancestor cancel surfaces `Cancelled` directly
from `.start()` on trio 0.29 — the lossy RTE requires
the child to absorb its cancel pre-`.started()`, which
is what `modden`'s `open_from_wks` teardown did. Trio's
nursery-exit wait defers cancel delivery to children,
so all tested shapes are deterministic (0 flakes / 50
runs).
Mutation verification:
```
mutation 1 (checkpoint_if_cancelled removed):
FAILED test_sibling_err_not_masked_by_startup_rte[True]
FAILED test_pure_oob_cancel_not_morphed_to_rte[True]
mutation 2 (guard relaxed to 'started' substring,
isinstance dropped):
FAILED test_childs_own_rte_never_demoted_to_cancel[never got started!]
FAILED test_childs_own_rte_never_demoted_to_cancel[1234]
```
Final runs:
```
tests/trionics/test_taskc.py: 9 passed in 0.03s
hammer: 0/50 runs failed
tests/trionics/ + tests/test_trioisms.py:
23 passed, 5 xfailed in 3.02s
```

View File

@ -0,0 +1,39 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: moc-teardown-completion-20260804
timestamp: 2026-08-04T03:03:09Z
git_ref: 65bf9df5
scope: code
substantive: true
raw_file: 20260804T030309Z_65bf9df5_prompt_io.raw.md
---
## Prompt
Patch Tractor's `maybe_open_context()` so the final consumer waits for
resource `__aexit__()` completion and receives cleanup errors. Reuse
`outcome.Outcome` for the exit result; use the smaller mutable
`_CtxExit` holder if that makes the implementation simpler. Add and run
the relevant existing unit tests, but do not commit or push the patch.
After reviewing the result, simplify `_CtxExit` back to an optional
exception because the success outcome carries no useful value.
## Response summary
Added an exception-backed completion handshake between
`_Cache.run_ctx()` and the final `maybe_open_context()` consumer.
Serialized consumer registration and final teardown under the per-key
lock, preserving that lock for queued entrants. Added deterministic
regressions for normal exit, cleanup errors, cancellation interactions,
service-nursery cancellation, and teardown re-entry.
## Files changed
- `tractor/trionics/_mngrs.py` - publish and unwrap cached exit outcomes.
- `tests/test_resource_cache.py` - cover completion and cancellation.
## Human edits
The user directed the final simplification from `outcome.Outcome` to an
optional exception field. The patch remains uncommitted.

View File

@ -0,0 +1,36 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-04T03:03:09Z
git_ref: 65bf9df5
diff_cmd: git diff HEAD~1..HEAD
---
Implemented cached-context exit completion in
`tractor.trionics.maybe_open_context()`.
> `git diff HEAD~1..HEAD -- tractor/trionics/_mngrs.py`
The generated implementation adds `_CtxExit`, whose `done` event
publishes an `outcome.Outcome[None]`. `_Cache.run_ctx()` records either
`Value(None)` or `Error(exc)` after the resource exit attempt. The final
MOC consumer signals `no_more_users`, waits for completion under a
shielded cancel scope, removes the per-key lock, and unwraps the outcome
so ordinary cleanup failures are raised at the consumer boundary.
`trio.Cancelled`, `KeyboardInterrupt`, and `SystemExit` continue through
the service task rather than being converted into regular cleanup
errors.
> `git diff HEAD~1..HEAD -- tests/test_resource_cache.py`
The generated regressions cover successful exit blocking, cleanup-error
delivery, final-user cancellation, cancellation combined with a cleanup
error, and service-nursery cancellation. The existing teardown re-entry
test now uses explicit events instead of a ten-second cleanup sleep and
asserts that the replacement resource is a fresh cache miss.
Verification:
`env PYTHONPATH="$PWD" /home/goodboy/repos/tractor/py313/bin/python -m pytest tests/test_resource_cache.py`
Result: `16 passed in 8.37s`.

View File

@ -0,0 +1,22 @@
# AI Prompt I/O Log - OpenCode
This directory tracks prompt inputs and model outputs for AI-assisted
development using `opencode`.
## Policy
Prompt logging follows the [NLNet generative AI policy][nlnet-ai]. All
substantive AI contributions are logged with:
- Model name and version
- Timestamps
- The prompts that produced the output
- Unedited model output (`.raw.md` files)
[nlnet-ai]: https://nlnet.nl/foundation/policies/generativeAI/
## Usage
Entries are created by the prompt-io workflow. Human contributors remain
accountable for all decisions. AI-generated content is never presented as
human-authored work.

View File

@ -2,16 +2,19 @@
`tractor.log`-wrapping unit tests. `tractor.log`-wrapping unit tests.
''' '''
import logging
from pathlib import Path from pathlib import Path
import shutil import shutil
from types import ModuleType from types import ModuleType
import pytest import pytest
import tractor import tractor
import trio
from tractor import ( from tractor import (
_code_load, _code_load,
log, log,
) )
from tractor.ipc import _chan
def test_root_pkg_not_duplicated_in_logger_name(): def test_root_pkg_not_duplicated_in_logger_name():
@ -222,6 +225,88 @@ def test_add_log_level_pluggable():
delattr(log.StackLevelAdapter, name.lower()) delattr(log.StackLevelAdapter, name.lower())
@pytest.mark.parametrize(
'suppression',
[
'level',
'logger',
'global',
],
)
def test_log_guard_skips_payload_formatting(
monkeypatch: pytest.MonkeyPatch,
suppression: str,
):
'''
Suppressed transport logs must not render payloads.
The original hot-path guard compared only the effective logger
level. A logger disabled through its `Logger.disabled` flag or
the global `logging.disable()` threshold could therefore still
call `pformat()` before `Logger.isEnabledFor()` discarded the
record.
Exercise effective-level, per-logger, and global suppression
independently. A poisoned `_chan.pformat()` proves rendering is
skipped, while the fake transport proves `Channel.send()` still
transmits the original payload and traceback-hiding flag.
'''
sent: list[tuple[object, bool]] = []
class FakeTransport:
async def send(
self,
payload: object,
hide_tb: bool = False,
) -> None:
sent.append((payload, hide_tb))
def fail_pformat(payload: object) -> str:
raise AssertionError(
f'suppressed log rendered payload: {payload!r}'
)
chan_log = log.get_logger(
name=f'guard_test.{suppression}',
)
std_log = chan_log.logger
orig_level: int = std_log.level
orig_disable: int = logging.root.manager.disable
transport_level: int = log.CUSTOM_LEVELS['TRANSPORT']
monkeypatch.setattr(_chan, 'log', chan_log)
monkeypatch.setattr(_chan, 'pformat', fail_pformat)
try:
logging.disable(logging.NOTSET)
std_log.setLevel(transport_level)
if suppression == 'level':
std_log.setLevel(logging.INFO)
elif suppression == 'logger':
monkeypatch.setattr(std_log, 'disabled', True)
else:
logging.disable(logging.CRITICAL)
assert not chan_log.isEnabledFor(transport_level)
transport = FakeTransport()
chan = _chan.Channel(transport=transport)
payload = object()
async def send_payload() -> None:
await chan.send(
payload,
hide_tb=True,
)
trio.run(send_payload)
assert sent == [(payload, True)]
finally:
std_log.setLevel(orig_level)
logging.disable(orig_disable)
# TODO, moar tests against existing feats: # TODO, moar tests against existing feats:
# ------ - ------ # ------ - ------
# - [ ] color settings? # - [ ] color settings?

View File

@ -9,6 +9,7 @@ from typing import Awaitable
import pytest import pytest
import trio import trio
from trio.testing import wait_all_tasks_blocked
import tractor import tractor
from tractor.trionics import ( from tractor.trionics import (
maybe_open_context, maybe_open_context,
@ -94,6 +95,232 @@ def test_resource_only_entered_once(key_on):
trio.run(main) trio.run(main)
def test_last_moc_user_waits_for_resource_exit():
'''
Verify the final user cannot return before resource teardown.
Previously the final `maybe_open_context()` user only signalled
`_Cache.run_ctx()` through its `no_more_users` event. The user
then returned while the service task was still running the
resource's `__aexit__()`, so callers could observe stale external
state immediately after their `async with` block.
The resource sets `exit_started` before blocking on
`allow_exit`. The user task must remain inside MOC until the test
releases that deterministic checkpoint and `__aexit__()` sets
`exit_finished`.
'''
async def main():
exit_started = trio.Event()
allow_exit = trio.Event()
exit_finished = trio.Event()
user_returned = trio.Event()
@acm
async def open_resource():
try:
yield
finally:
exit_started.set()
await allow_exit.wait()
exit_finished.set()
async def use_resource():
async with maybe_open_context(open_resource):
pass
assert exit_finished.is_set()
user_returned.set()
async with (
tractor.open_root_actor(),
trio.open_nursery() as tn,
):
tn.start_soon(use_resource)
await exit_started.wait()
assert not user_returned.is_set()
allow_exit.set()
await user_returned.wait()
trio.run(main)
def test_moc_delivers_resource_exit_error():
'''
Verify a resource exit error reaches the final MOC user.
Previously `_Cache.run_ctx()` executed the cached resource's
`__aexit__()` after the final user had returned. An exit failure
therefore surfaced later through the actor service nursery rather
than at the user's `async with maybe_open_context()` boundary.
This resource raises a unique `ResourceExitError` during exit.
Catching that exact instance around MOC proves the service task
delivered the failure to the final user without replacing it.
'''
class ResourceExitError(Exception):
pass
exit_error = ResourceExitError('resource exit failed')
async def main():
@acm
async def open_resource():
yield
raise exit_error
async with tractor.open_root_actor():
with pytest.raises(ResourceExitError) as exc_info:
async with maybe_open_context(open_resource):
pass
assert exc_info.value is exit_error
trio.run(main)
def test_moc_final_user_cancellation_waits_for_exit():
'''
Verify final-user cancellation still waits for successful exit.
Previously cancellation escaped the final MOC user immediately
after it signalled `_Cache.run_ctx()`, leaving resource exit to
finish later in the actor service task. This violated the context
manager boundary even when cleanup itself succeeded.
The consumer cancels its own scope while holding the sole cached
resource. The resource sets `exit_finished` from its `finally`
block, and the consumer checks that event immediately after its
cancel scope catches `trio.Cancelled`. This proves MOC's
completion wait is shielded without suppressing the original
cancellation.
'''
async def main():
exit_finished = trio.Event()
@acm
async def open_resource():
try:
yield
finally:
exit_finished.set()
async with tractor.open_root_actor():
with trio.CancelScope() as cs:
async with maybe_open_context(open_resource):
cs.cancel()
await trio.sleep_forever()
assert cs.cancelled_caught
assert exit_finished.is_set()
trio.run(main)
def test_moc_exit_error_masks_final_user_cancellation():
'''
Verify cleanup errors survive final-user cancellation.
A cancelled final user previously signalled `no_more_users` and
propagated `trio.Cancelled` before `_Cache.run_ctx()` completed
resource exit. If `__aexit__()` then failed, its error was
detached from the API call which caused teardown.
The consumer cancels its own scope at a deterministic checkpoint
inside MOC. Resource exit raises `ResourceExitError`; observing
that exact error outside the cancel scope proves MOC shields the
completion wait and applies normal context-manager masking, where
a cleanup failure replaces the active cancellation.
'''
class ResourceExitError(Exception):
pass
exit_error = ResourceExitError('resource exit failed')
async def main():
@acm
async def open_resource():
yield
raise exit_error
async with tractor.open_root_actor():
with pytest.raises(ResourceExitError) as exc_info:
with trio.CancelScope() as cs:
async with maybe_open_context(open_resource):
cs.cancel()
await trio.sleep_forever()
assert exc_info.value is exit_error
trio.run(main)
def test_moc_service_nursery_cancellation_completes_exit():
'''
Verify service-nursery cancellation cannot strand a final user.
`_Cache.run_ctx()` and an MOC consumer may share a
caller-provided service nursery. Cancelling that nursery
interrupts the service task's `no_more_users` wait and the
consumer body together. A shielded final-user wait would deadlock
if `run_ctx()` failed to publish completion while propagating its
own `trio.Cancelled`.
The outer task waits for resource entry, then cancels the exact
nursery containing both tasks. The resource shields one cleanup
checkpoint and sets `exit_finished`; observing both it and
`service_finished` proves cancellation propagated normally while
MOC's completion handshake terminated deterministically.
'''
async def main():
resource_entered = trio.Event()
exit_finished = trio.Event()
service_finished = trio.Event()
service_tn: trio.Nursery|None = None
@acm
async def open_resource():
try:
resource_entered.set()
yield
finally:
with trio.CancelScope(shield=True):
await trio.lowlevel.checkpoint()
exit_finished.set()
async def use_resource(tn: trio.Nursery):
async with maybe_open_context(
open_resource,
tn=tn,
):
await trio.sleep_forever()
async def run_service():
nonlocal service_tn
async with trio.open_nursery() as tn:
service_tn = tn
tn.start_soon(use_resource, tn)
service_finished.set()
async with trio.open_nursery() as outer_tn:
outer_tn.start_soon(run_service)
await resource_entered.wait()
assert service_tn is not None
service_tn.cancel_scope.cancel()
await service_finished.wait()
assert exit_finished.is_set()
trio.run(main)
@tractor.context @tractor.context
async def streamer( async def streamer(
ctx: tractor.Context, ctx: tractor.Context,
@ -548,43 +775,52 @@ def test_moc_reentry_during_teardown(
loglevel: str, loglevel: str,
): ):
''' '''
Reproduce the piker `open_cached_client('kraken')` race: Reproduce re-entry while an identical cached context exits.
- same `acm_func`, NO kwargs (identical `ctx_key`) - multiple tasks use the same `acm_func` with no kwargs,
- multiple tasks share the cached resource producing an identical `ctx_key`;
- all users exit -> teardown starts - all users leave and the final user starts resource teardown;
- a NEW task enters during `_Cache.run_ctx.__aexit__` - `_Cache.run_ctx()` removes the cached value and resource entry
- `values[ctx_key]` is gone (popped in inner finally) before entering the resource's blocking `__aexit__()` body;
but `resources[ctx_key]` still exists (outer finally - a new task attempts to enter that same `ctx_key` during exit;
hasn't run yet bc the acm cleanup has checkpoints) - the per-key lock keeps that entrant queued until exit
- old code: `assert not resources.get(ctx_key)` FIRES completes;
- the entrant then receives a fresh cache miss and resource.
This models the real-world scenario where `brokerd.kraken` Without teardown sharing the registration lock, re-entry could
tasks concurrently call `open_cached_client('kraken')` race resource replacement while the prior generation was still
(same `acm_func`, empty kwargs, shared `ctx_key`) and exiting. The final user could also return before that exit
the teardown/re-entry race triggers intermittently. completed.
The first resource generation signals `in_aexit` and waits on
`allow_aexit`. The re-entry task signals `reentry_started` and
blocks inside MOC; only after `wait_all_tasks_blocked()` confirms
that ordering does the coordinator release cleanup. The entrant
must then receive a fresh cache miss. `first_done` additionally
proves the first MOC user observed completed teardown before
returning.
''' '''
async def main(): async def main():
in_aexit = trio.Event() in_aexit = trio.Event()
allow_aexit = trio.Event()
reentry_started = trio.Event()
generation: int = 0
@acm @acm
async def cached_client(): async def cached_client():
''' '''
Simulates `kraken.api.get_client()`: Simulate a no-argument `kraken.api.get_client()`.
- no params (all callers share one `ctx_key`)
- slow-ish cleanup to widen the race window
between `values.pop()` and `resources.pop()`
inside `_Cache.run_ctx`.
''' '''
nonlocal generation
generation += 1
resource_generation: int = generation
yield 'the-client' yield 'the-client'
# Signal that we're in __aexit__ — at this if resource_generation == 1:
# point `values` has already been popped by
# `run_ctx`'s inner finally, but `resources`
# is still alive (outer finally hasn't run).
in_aexit.set() in_aexit.set()
await trio.sleep(10) await allow_aexit.wait()
first_done = trio.Event() first_done = trio.Event()
@ -598,16 +834,25 @@ def test_moc_reentry_during_teardown(
async def reenter_during_teardown(): async def reenter_during_teardown():
''' '''
Wait for the acm's `__aexit__` to start (meaning Wait for the acm's `__aexit__` to start (meaning
`values` is popped but `resources` still exists), the cached value is no longer available), then re-enter.
then re-enter triggering the assert.
''' '''
await in_aexit.wait() await in_aexit.wait()
# Tell the coordinator this task is about to enter MOC.
# `Event.set()` is not a checkpoint. Though `async with`
# awaits MOC's `__aenter__()`, its async generator runs
# synchronously until the held per-key `lock.acquire()`
# actually suspends this task.
reentry_started.set()
async with maybe_open_context( async with maybe_open_context(
cached_client, cached_client,
) as (cache_hit, value): ) as (cache_hit, value):
assert not cache_hit
assert value == 'the-client' assert value == 'the-client'
await first_done.wait()
with trio.fail_after(5): with trio.fail_after(5):
async with ( async with (
tractor.open_root_actor( tractor.open_root_actor(
@ -619,5 +864,15 @@ def test_moc_reentry_during_teardown(
): ):
tn.start_soon(use_and_exit) tn.start_soon(use_and_exit)
tn.start_soon(reenter_during_teardown) tn.start_soon(reenter_during_teardown)
await reentry_started.wait()
# Wait until the re-entry task is queued on MOC's
# per-key lock while `_Cache.run_ctx()` remains
# blocked in the first generation's `__aexit__()`.
# Only then release cleanup, making the intended
# enter-during-sibling-exit ordering deterministic.
await wait_all_tasks_blocked()
assert not first_done.is_set()
allow_aexit.set()
trio.run(main) trio.run(main)

View File

@ -0,0 +1,312 @@
'''
`tractor.trionics._taskc.start_or_cancel()` unit tests.
`trio.Nursery.start()` collapses an out-of-band (ancestor)
cancellation into a lossy,
`RuntimeError('child exited without calling
task_status.started()')`
whenever the started child exits pre-`.started()` WITHOUT
propagating the ambient `trio.Cancelled`; a common outcome
when the child (or any lib code it calls) runs a graceful
teardown which absorbs the cancel and returns early. Our
`start_or_cancel()` wrapper re-surfaces the real in-flight
cancellation in that case so the true root error/cancel
propagates to the `.start()` caller instead.
These tests verify both that repair AND document upstream
`trio`'s current lossy behaviour via the
`use_start_or_cancel=False` parametrizations; if a `trio`
upgrade breaks one of THOSE cases it likely means upstream
shipped better startup-cancellation porcelain and our
wrapper deserves a re-audit!
The core use case was dug out of `modden`'s
`progman.open_wks()` program-spawn machinery as per gh
issue #474; the wrapper landed originally via gh PR #464.
'''
import pytest
import trio
from trio import TaskStatus
from tractor.trionics import start_or_cancel
async def absorbs_cancel_pre_started(
task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
):
'''
Swallow the ambient (ancestor-scope) cancel and return
early, a naughty-but-realistic graceful-teardown pattern
and the exact shape which causes `trio.Nursery.start()`
to raise its lossy startup `RuntimeError` in place of
the real `trio.Cancelled`.
'''
try:
await trio.sleep(2)
except trio.Cancelled:
return
task_status.started()
async def raise_val_err():
'''
Sibling task which blows up (fast) thus OOB-cancelling
the shared parent-nursery's cancel-scope.
'''
await trio.lowlevel.checkpoint()
raise ValueError('sibling blew up!')
@pytest.mark.parametrize(
'use_start_or_cancel',
[
True,
False,
],
)
def test_sibling_err_not_masked_by_startup_rte(
use_start_or_cancel: bool,
):
'''
The `modden.runtime.progman` use case: a sibling task
errors while the `.start()`-ed child is still
pre-`.started()`, OOB-cancelling the shared nursery
scope; the child absorbs its cancel (graceful teardown)
and exits early.
- with `start_or_cancel()` the in-flight cancellation
is re-surfaced as the real `trio.Cancelled` (then
absorbed by the cancelled nursery scope) so ONLY the
root-cause sibling error escapes the nursery.
- with a bare `.start()`, upstream `trio` (currently)
also delivers its lossy startup `RuntimeError`
alongside, obscuring that the child was in fact
cancelled due to the sibling's error.
'''
async def main():
async with trio.open_nursery() as tn:
tn.start_soon(raise_val_err)
if use_start_or_cancel:
await start_or_cancel(
tn,
absorbs_cancel_pre_started,
)
else:
await tn.start(absorbs_cancel_pre_started)
with pytest.raises(ExceptionGroup) as excinfo:
trio.run(main)
eg: ExceptionGroup = excinfo.value
val_eg, rest_eg = eg.split(ValueError)
assert len(val_eg.exceptions) == 1
if use_start_or_cancel:
# the re-surfaced `Cancelled` is absorbed by the
# (sibling-error cancelled) nursery scope leaving
# NO startup-noise, just the root cause.
assert rest_eg is None
else:
# the `trio` wart: a lossy startup RTE rides along
# with (and distracts from) the root cause.
rte = rest_eg.exceptions[0]
assert isinstance(rte, RuntimeError)
assert 'child exited without calling' in rte.args[0]
@pytest.mark.parametrize(
'use_start_or_cancel',
[
True,
False,
],
)
def test_pure_oob_cancel_not_morphed_to_rte(
use_start_or_cancel: bool,
):
'''
A plain (error-free) ancestor `CancelScope.cancel()`
fired while the (cancel-absorbing) child is still
pre-`.started()`:
- `start_or_cancel()` re-surfaces the `Cancelled` so
the cancelled scope exits CLEAN, no error at all.
- a bare `.start()` (currently) morphs the plain
cancel into an (eg-wrapped) startup `RuntimeError`.
'''
async def main():
with trio.CancelScope() as cs:
async with trio.open_nursery() as tn:
async def canceller():
await trio.lowlevel.checkpoint()
cs.cancel()
tn.start_soon(canceller)
if use_start_or_cancel:
await start_or_cancel(
tn,
absorbs_cancel_pre_started,
)
else:
await tn.start(
absorbs_cancel_pre_started,
)
assert cs.cancelled_caught
if use_start_or_cancel:
trio.run(main)
else:
with pytest.raises(ExceptionGroup) as excinfo:
trio.run(main)
rte = excinfo.value.exceptions[0]
assert isinstance(rte, RuntimeError)
assert 'child exited without calling' in rte.args[0]
@pytest.mark.parametrize(
'use_start_or_cancel',
[
True,
False,
],
)
def test_genuine_startup_rte_still_raised(
use_start_or_cancel: bool,
):
'''
Absent ANY in-flight cancellation, a child exiting
cleanly without calling `task_status.started()` is a
genuine startup-protocol bug; `start_or_cancel()` must
re-raise the resulting `RuntimeError` exactly like a
bare `.start()` does.
'''
async def exits_wo_started(
task_status: TaskStatus[None] = (
trio.TASK_STATUS_IGNORED
),
):
await trio.lowlevel.checkpoint()
async def main():
async with trio.open_nursery() as tn:
with pytest.raises(RuntimeError) as excinfo:
if use_start_or_cancel:
await start_or_cancel(
tn,
exits_wo_started,
)
else:
await tn.start(exits_wo_started)
rte = excinfo.value
assert (
'child exited without calling'
in
rte.args[0]
)
trio.run(main)
@pytest.mark.parametrize(
'rte_arg',
[
# a bare `'started' in args[0]` substring match
# would (wrongly) demote this one to a `Cancelled`
# under ambient cancellation.
'never got started!',
# non-`str` first-arg edge; must not `TypeError`
# inside the wrapper's msg-match guard.
1234,
],
)
def test_childs_own_rte_never_demoted_to_cancel(
rte_arg: str|int,
):
'''
A child's OWN `RuntimeError`, one which merely smells
like `trio`'s startup wording (or carries a non-`str`
first arg), raised under ambient cancellation must NOT
be demoted to a `trio.Cancelled` by the exact-msg-match
guard inside `start_or_cancel()`; the real error must
always propagate to the caller unchanged.
'''
async def cancels_cs_then_raises(
task_status: TaskStatus[None] = (
trio.TASK_STATUS_IGNORED
),
):
# cancel the ambient (ancestor) scope then raise
# sync-ly, no checkpoint between, so the child
# deterministically dies with ITS error while the
# caller is under effective cancellation.
cs.cancel()
raise RuntimeError(rte_arg)
cs = trio.CancelScope()
async def main():
with cs:
async with trio.open_nursery() as tn:
await start_or_cancel(
tn,
cancels_cs_then_raises,
)
with pytest.raises(ExceptionGroup) as excinfo:
trio.run(main)
rte = excinfo.value.exceptions[0]
assert isinstance(rte, RuntimeError)
assert rte.args[0] == rte_arg
def test_started_value_and_args_passthru():
'''
Happy path: positional args, the `name=` kwarg and the
`.started(value)`-delivered value all pass through
`start_or_cancel()` identically to a bare `.start()`.
'''
async def echo_started(
*args,
task_status: TaskStatus[tuple] = (
trio.TASK_STATUS_IGNORED
),
):
task_name: str = trio.lowlevel.current_task().name
task_status.started((
args,
task_name,
))
async def main():
async with trio.open_nursery() as tn:
(
args,
task_name,
) = await start_or_cancel(
tn,
echo_started,
'chillin',
10,
name='doggy',
)
assert args == ('chillin', 10)
assert task_name == 'doggy'
trio.run(main)

View File

@ -198,9 +198,6 @@ class Channel:
# assert transport.raddr == addr # assert transport.raddr == addr
chan = Channel(transport=transport) chan = Channel(transport=transport)
# ?TODO, compact this into adapter level-methods?
# -[ ] would avoid extra repr-calcs if level not active?
# |_ how would the `calc_if_level` look though? func?
if log.at_least_level('runtime'): if log.at_least_level('runtime'):
from tractor.devx import ( from tractor.devx import (
pformat as _pformat, pformat as _pformat,
@ -325,6 +322,8 @@ class Channel:
''' '''
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
try: try:
if log.at_least_level('transport'):
# don't materialize the payload repr if not necessary
log.transport( log.transport(
'=> send IPC msg:\n\n' '=> send IPC msg:\n\n'
f'{pformat(payload)}\n' f'{pformat(payload)}\n'

View File

@ -309,7 +309,10 @@ class MsgpackTransport(MsgTransport):
log.transport(f'received header {size}') # type: ignore log.transport(f'received header {size}') # type: ignore
msg_bytes: bytes = await self.recv_stream.receive_exactly(size) msg_bytes: bytes = await self.recv_stream.receive_exactly(size)
log.transport(f"received {msg_bytes}") # type: ignore if log.at_least_level('transport'):
log.transport( # type: ignore
f'received {msg_bytes}'
)
try: try:
# NOTE: lookup the `trio.Task.context`'s var for # NOTE: lookup the `trio.Task.context`'s var for
# the current `MsgCodec`. # the current `MsgCodec`.

View File

@ -111,9 +111,7 @@ def at_least_level(
if isinstance(level, str): if isinstance(level, str):
level: int = CUSTOM_LEVELS[level.upper()] level: int = CUSTOM_LEVELS[level.upper()]
if log.getEffectiveLevel() <= level: return log.isEnabledFor(level)
return True
return False
# TODO, compare with using a "filter" instead? # TODO, compare with using a "filter" instead?

View File

@ -306,12 +306,12 @@ class PldRx(Struct):
): ):
try: try:
pld: PayloadT = self._pld_dec.decode(pld) pld: PayloadT = self._pld_dec.decode(pld)
if log.at_least_level('runtime'):
# don't materialize the payload repr if not necessary
log.runtime( log.runtime(
f'Decoded payload for\n' f'Decoded payload for\n'
# f'\n' f'\n'
f'{msg}\n' f'{msg}\n'
# ^TODO?, ideally just render with `,
# pld={decode}` in the `msg.pformat()`??
f'where, ' f'where, '
f'{type(msg).__name__}.pld={pld!r}\n' f'{type(msg).__name__}.pld={pld!r}\n'
) )

View File

@ -1003,17 +1003,15 @@ async def process_messages(
task_status.started(loop_cs) task_status.started(loop_cs)
async for msg in chan: async for msg in chan:
if log.at_least_level('transport'):
log.transport( # type: ignore log.transport( # type: ignore
f'IPC msg from peer\n' f'IPC msg from peer\n'
f'<= {chan.aid.reprol()}\n\n' f'<= {chan.aid.reprol()}\n\n'
# TODO: use of the pprinting of structs is # TODO: pretty-printing structs is FRAGILE;
# FRAGILE and should prolly not be # -[ ] add a non-raising log formatter with
# # native-repr fallback before using
# avoid fmting depending on loglevel for perf? # `.msg.pretty_struct` here.
# -[ ] specifically `pretty_struct.pformat()` sub-call..?
# - how to only log-level-aware actually call this?
# -[ ] use `.msg.pretty_struct` here now instead!
# f'{pretty_struct.pformat(msg)}\n' # f'{pretty_struct.pformat(msg)}\n'
f'{msg}\n' f'{msg}\n'
) )
@ -1262,6 +1260,7 @@ async def process_messages(
log.exception(message) log.exception(message)
raise RuntimeError(message) raise RuntimeError(message)
if log.at_least_level('transport'):
log.transport( log.transport(
'Waiting on next IPC msg from\n' 'Waiting on next IPC msg from\n'
f'peer: {chan.aid.reprol()}\n' f'peer: {chan.aid.reprol()}\n'

View File

@ -198,6 +198,16 @@ async def gather_contexts(
# Further potential examples of interest: # Further potential examples of interest:
# https://gist.github.com/njsmith/cf6fc0a97f53865f2c671659c88c1798#file-cache-py-L8 # https://gist.github.com/njsmith/cf6fc0a97f53865f2c671659c88c1798#file-cache-py-L8
class _CtxExit:
'''
Completion state for a cached context's shared exit.
'''
def __init__(self) -> None:
self.done = trio.Event()
self.error: Exception|None = None
class _Cache: class _Cache:
''' '''
Globally (actor-processs scoped) cached, task access to Globally (actor-processs scoped) cached, task access to
@ -213,7 +223,11 @@ class _Cache:
values: dict[Any, Any] = {} values: dict[Any, Any] = {}
resources: dict[ resources: dict[
Hashable, Hashable,
tuple[trio.Nursery, trio.Event] tuple[
trio.Nursery,
trio.Event,
_CtxExit,
],
] = {} ] = {}
# nurseries: dict[int, trio.Nursery] = {} # nurseries: dict[int, trio.Nursery] = {}
no_more_users: trio.Event|None = None no_more_users: trio.Event|None = None
@ -223,19 +237,39 @@ class _Cache:
cls, cls,
mng, mng,
ctx_key: tuple, ctx_key: tuple,
ctx_exit: _CtxExit,
task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED, task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED,
) -> None: ) -> None:
entered: bool = False
try:
async with mng as value: async with mng as value:
_, no_more_users = cls.resources[ctx_key] entered = True
(
_,
no_more_users,
_,
) = cls.resources[ctx_key]
cls.values[ctx_key] = value cls.values[ctx_key] = value
task_status.started(value) task_status.started(value)
try: try:
await no_more_users.wait() await no_more_users.wait()
finally: finally:
value = cls.values.pop(ctx_key) cls.values.pop(ctx_key)
cls.resources.pop(ctx_key) cls.resources.pop(ctx_key)
except Exception as exc:
if not entered:
raise
# Deliver regular `__aexit__()` failures to the final
# consumer instead of raising into the service nursery.
ctx_exit.error = exc
finally:
if entered:
ctx_exit.done.set()
class _UnresolvedCtx: class _UnresolvedCtx:
''' '''
@ -281,9 +315,10 @@ async def maybe_open_context(
) )
# yielded output # yielded output
# sentinel = object()
yielded: Any = _UnresolvedCtx yielded: Any = _UnresolvedCtx
user_registered: bool = False user_registered: bool = False
ctx_exit: _CtxExit|None = None
exit_error: Exception|None = None
# Lock resource acquisition around task racing / ``trio``'s # Lock resource acquisition around task racing / ``trio``'s
# scheduler protocol. # scheduler protocol.
@ -300,7 +335,6 @@ async def maybe_open_context(
] = trio.StrictFIFOLock() ] = trio.StrictFIFOLock()
header: str = 'Allocated NEW lock for @acm_func,\n' header: str = 'Allocated NEW lock for @acm_func,\n'
else: else:
await trio.lowlevel.checkpoint()
header: str = 'Reusing OLD lock for @acm_func,\n' header: str = 'Reusing OLD lock for @acm_func,\n'
log.debug( log.debug(
@ -368,7 +402,6 @@ async def maybe_open_context(
resources = _Cache.resources resources = _Cache.resources
entry: tuple|None = resources.get(ctx_key) entry: tuple|None = resources.get(ctx_key)
if entry: if entry:
service_tn, ev = entry
raise RuntimeError( raise RuntimeError(
f'Caching resources ALREADY exist?!\n' f'Caching resources ALREADY exist?!\n'
f'ctx_key={ctx_key!r}\n' f'ctx_key={ctx_key!r}\n'
@ -376,12 +409,18 @@ async def maybe_open_context(
f'task: {task}\n' f'task: {task}\n'
) )
resources[ctx_key] = (service_tn, trio.Event()) ctx_exit = _CtxExit()
resources[ctx_key] = (
service_tn,
trio.Event(),
ctx_exit,
)
try: try:
yielded: Any = await service_tn.start( yielded: Any = await service_tn.start(
_Cache.run_ctx, _Cache.run_ctx,
mngr, mngr,
ctx_key, ctx_key,
ctx_exit,
) )
except BaseException: except BaseException:
# If `run_ctx` (wrapping the acm's `__aenter__`) # If `run_ctx` (wrapping the acm's `__aenter__`)
@ -427,6 +466,11 @@ async def maybe_open_context(
raise taskc raise taskc
else: else:
# XXX, cached-entry-path # XXX, cached-entry-path
(
_,
_,
ctx_exit,
) = _Cache.resources[ctx_key]
_Cache.users[ctx_key] += 1 _Cache.users[ctx_key] += 1
user_registered = True user_registered = True
log.debug( log.debug(
@ -445,24 +489,17 @@ async def maybe_open_context(
) )
finally: finally:
if lock.locked():
stats: trio.LockStatistics = lock.statistics()
owner: trio.Task|None = stats.owner
log.error(
f'Lock never released by last owner={owner!r} !?\n'
f'{stats}\n'
f'\n'
f'task={task!r}\n'
f'ctx_key={ctx_key!r}\n'
f'acm_func={acm_func}\n'
)
if user_registered: if user_registered:
# Serialize user registration and teardown under the same
# per-key lock so no entrant can acquire a resource after
# its final user has committed to exiting it.
with trio.CancelScope(shield=True):
await lock.acquire()
try:
_Cache.users[ctx_key] -= 1 _Cache.users[ctx_key] -= 1
if yielded is not _UnresolvedCtx: # If no consumers remain, keep entrants queued
# if no more consumers, teardown the client # until the cached context has completely exited.
if _Cache.users[ctx_key] <= 0: if _Cache.users[ctx_key] <= 0:
log.debug( log.debug(
f'De-allocating @acm-func entry\n' f'De-allocating @acm-func entry\n'
@ -470,19 +507,39 @@ async def maybe_open_context(
f'acm_func={acm_func!r}\n' f'acm_func={acm_func!r}\n'
) )
# XXX: if we're cancelled we the entry may have never # XXX: if we're cancelled, the entry may
# been entered since the nursery task was killed. # have never been entered since the nursery
# _, no_more_users = _Cache.resources[ctx_key] # task was killed.
entry = _Cache.resources.get(ctx_key) entry = _Cache.resources.get(ctx_key)
if entry: if entry:
_, no_more_users = entry (
_,
no_more_users,
ctx_exit,
) = entry
no_more_users.set() no_more_users.set()
maybe_lock = _Cache.locks.pop( assert ctx_exit is not None
ctx_key, await ctx_exit.done.wait()
None, exit_error = ctx_exit.error
)
if maybe_lock is None: # A queued entrant already holds a reference
# to this lock. Keep it registered until that
# task has acquired and released it.
stats = lock.statistics()
if not stats.tasks_waiting:
maybe_lock = _Cache.locks.get(ctx_key)
if maybe_lock is lock:
_Cache.locks.pop(ctx_key)
else:
log.error( log.error(
f'Resource lock for {ctx_key} ALREADY POPPED?' f'Resource lock for {ctx_key} '
f'was replaced before teardown?'
) )
finally:
lock.release()
if exit_error is not None:
# Always re-raise a regular `__aexit__()` error at the
# final consumer's context boundary.
raise exit_error