diff --git a/.agents/skills/duroxide-python-orchestrations/SKILL.md b/.agents/skills/duroxide-python-orchestrations/SKILL.md index 45328c7..86ddb21 100644 --- a/.agents/skills/duroxide-python-orchestrations/SKILL.md +++ b/.agents/skills/duroxide-python-orchestrations/SKILL.md @@ -273,7 +273,7 @@ pytest tests/scenarios/ -v # scenario tests (6 tests) - Use `SqliteProvider.in_memory()` for fast isolated tests (SQLite smoketest only) - All PG tests need `DATABASE_URL` in `.env` (loaded by `python-dotenv`) - Each test file uses a separate PG schema for isolation -- Use short `runtime.shutdown(100)` timeout — it waits the full duration +- `runtime.shutdown(100)` permits 100 ms of grace plus five seconds of cleanup headroom. It can finish early; timeout raises while cleanup remains owned and requires application/supervisor termination. - Set `RUST_LOG=info` and use `pytest -s` to see traces in test output - Use `worker_lock_timeout_ms=2000` in tests needing fast activity cancellation detection diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d08803c..b3fe414 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -210,7 +210,7 @@ Orchestration code must be deterministic — same input + history = same sequenc - PostgreSQL tests use schema isolation (one schema per test file) - `python-dotenv` loads `DATABASE_URL` from `.env` - `PyRuntimeOptions(dispatcher_poll_interval_ms=50)` for fast test dispatch -- `runtime.shutdown(100)` — short timeout, it waits the full duration +- `runtime.shutdown(100)` — 100 ms grace plus five seconds total headroom; idle work can finish early, and failures raise while retaining cleanup - Use `SqliteProvider.in_memory()` only for SQLite smoketest - `worker_lock_timeout_ms=2000` for tests needing fast cancellation detection diff --git a/CHANGELOG.md b/CHANGELOG.md index 000e160..1e66333 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add string-valued `RuntimeStopStatus` and owned, metadata-only + `RuntimeStopObserver` through `Runtime.stop_observer()`. Synchronous change + waits release the GIL, support finite timeout and Python signals, and allow + terminal cross-thread close. Final state arrives before normal source end. + Observation does not retain worker callbacks or change shutdown ownership, + deadlines, and application policy. + +### Changed + +- Use fallible core startup and bounded shutdown waits while preserving successful `None` returns. + Total waiting is the supplied grace plus five seconds. + Omitted grace is one second. Idle workers can finish early. + Invalid durations fail before ownership changes, including for no-work and repeated calls. + Repeated calls retain the original operation, errors, and deadlines. + They do not report success merely because the wrapper transferred its core. + Registry and preparation failures become ordinary retained startup errors. + Registration policy does not change. +- Pre-start shutdown is terminal. + Start and registration calls fail after activation or stop. + Post-stop metrics remain unavailable. +- Shutdown timeout raises `TimeoutError`. Startup and operational failures raise `RuntimeError`. + Incomplete cleanup requires application- or supervisor-owned process termination, even after actual late cleanup. + Finite grace does not guarantee bounded forced cleanup. + This adapter does not add .NET's foreign-continuation barrier or own detached application work. + Release requires a matching published core dependency. + Do not distribute local development overrides. + +### Fixed + +- Release obsolete native activity-callback ownership after proven cleanup, + including quiescent failures and late completion without another shutdown call. + This breaks bound-method and self-capturing callback cycles that Python cannot otherwise collect. + The first budgets and terminal outcomes remain available. + Incomplete cleanup or a lost cleanup owner still retains ownership. + Independent Clients, providers, and the global orchestration-handler map do not change. + +### Tests + +- Add SQLite-only lifecycle coverage through the real wrapper. + Cases include feature-gated provider and fault controls, repeated ordered races, + isolated held-provider probes, and production-export checks. +- Activity-only collection controls, failed-startup metrics, rejected-registration + atomicity, and low-level driver preconditions. + Subprocess failures retain bounded diagnostics. + Transport watchdogs are separate from the unchanged shutdown timing limits. + Windows tests measure those limits with high-resolution clocks. + ## [0.2.2] - 2026-10-21 ### Changed diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3ee9c9f..ba66f3f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -29,4 +29,27 @@ cargo clippy --all-targets pytest -v ``` -After Rust source changes (`src/*.rs`), re-run `maturin develop` before running Python tests. \ No newline at end of file +After Rust source changes (`src/*.rs`), re-run `maturin develop` before running Python tests. + +### Lifecycle checks without PostgreSQL + +Use an isolated virtual environment with the declared development dependencies. +Rebuild and install the extension for each mode. +Importing an old extension is not lifecycle evidence. +For example, in PowerShell with that environment activated: + +```powershell +maturin develop --features test-hooks +$env:DUROXIDE_LIFECYCLE_TEST_HOOKS = '1' +python -m pytest -v tests\test_lifecycle.py tests\test_e2e.py::test_sqlite_smoketest +Remove-Item Env:DUROXIDE_LIFECYCLE_TEST_HOOKS +maturin develop +python -m pytest -v tests\test_lifecycle.py tests\test_e2e.py::test_sqlite_smoketest +``` + +Instrumented cases exercise real provider waits and contained faults in owned tasks. +The suite checks native import provenance and runs 100 repetitions per ordered race. +It also checks bounded child-process cleanup and confirms that production builds omit test exports. +Production mode skips only hook-dependent cases. +Do not publish instrumented assets or temporary local core overrides. +Release requires a matching published core minimum. \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index 63903c8..3ca3e0a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,9 @@ edition = "2021" name = "duroxide_python" crate-type = ["cdylib"] +[features] +test-hooks = ["duroxide/test-hooks"] + [dependencies] duroxide = { version = "0.1.32", features = ["sqlite"] } duroxide-pg = "0.1.37" diff --git a/README.md b/README.md index 45fa6f4..6b8f173 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,62 @@ Write durable workflows as Python generators. The Rust runtime handles replay, p - **Activity client access** — `ctx.get_client()` lets activities start new orchestrations - **Runtime metrics** — `metrics_snapshot()` for orchestration/activity counters +## Runtime lifecycle (unreleased) + +`runtime.start()` returns after startup. Core startup errors raise `RuntimeError`. +`runtime.shutdown(timeout_ms)` still returns `None` on success. +The argument is a nonnegative integer grace period in milliseconds. +It must fit the native monotonic deadline range. +Omitted grace is 1000 ms. Total waiting is grace plus 5000 ms. +Idle work can finish early. + +The first stop fixes both deadlines. Repeated calls observe the same operation. +They can report timeout, operational errors, or actual late completion. +Shutdown before start is terminal. Later start or registration calls fail. +Metrics remain unavailable after stop. +Serialize lifecycle and registration calls on each runtime. + +Timeout raises `TimeoutError`. Startup and operational failures raise `RuntimeError`. +A timeout means cleanup is still owned and incomplete. +Do not restart the worker. +The application or supervisor must terminate the process, even if a later observation reports completion. +The SDK does not terminate the process. + +This adapter does not track arbitrary Python work. +It does not add the .NET SDK's additional managed foreign-retirement barrier. +Independent provider and Client resources have their own lifetime. See the +[lifecycle guide](docs/user-guide.md#runtime-lifecycle) for validation and an +executable early-drain/repeated-stop example. +**Quiescence** means that tracked core execution and cleanup have ended. +Proven cleanup releases obsolete native activity-callback ownership, including after quiescent failures and late completion. +Terminal outcomes and the original budgets remain available. +Unproven cleanup never permits callback release. + +These unreleased changes require a compatible core lifecycle release before distribution. +A temporary local Cargo override provides development evidence, not a release dependency pin. + +### Passive stop observation + +`runtime.stop_observer()` creates an owned, metadata-only `RuntimeStopObserver`. +Read its `state` before calling `wait_for_change(last_seen, timeout_ms=None)`. +`RuntimeStopStatus` has string-valued `NOT_REQUESTED`, `REQUESTED`, and sticky +`FAULTED` members. They describe recorded stop/fault state, not readiness or +completed cleanup. A late observer can already see a fault. + +Use one pending wait per observer. A finite timeout raises `TimeoutError`; +`KeyboardInterrupt` propagates. Either permits retry without stopping the worker. +`close()` can run from another thread and is terminal and idempotent. +Pending and later operations fail after close. Normal source end instead +returns an unseen final state before `None`. + +The synchronous wait releases the GIL and checks Python signals. +It still blocks its calling thread. Do not run it on the asyncio event-loop +thread or assume that canceling `asyncio.to_thread` cancels the underlying wait. +Use a finite timeout or explicitly close the observer. +Observation retains no worker callback ownership and changes no shutdown +deadlines or termination requirements. See the +[observer example](docs/user-guide.md#passive-stop-observation). + ## Installation ```bash diff --git a/ci/smoke/smoke.py b/ci/smoke/smoke.py index 332b94b..8eb4fd1 100644 --- a/ci/smoke/smoke.py +++ b/ci/smoke/smoke.py @@ -18,15 +18,26 @@ """ import os +import hashlib +import json import platform import sys import tempfile import time +from pathlib import Path from duroxide import SqliteProvider, Client, Runtime, RuntimeOptions +from duroxide import _duroxide as native def main() -> int: + extension = Path(native.__file__).resolve() + assert extension.is_relative_to(Path(sys.prefix).resolve()), "smoke must load the installed wheel" + assert not hasattr(native, "_lifecycle_test_hooks") + assert not hasattr(native, "LifecycleTestHooks") + print("[smoke] native=" + json.dumps({ + "path": str(extension), "sha256": hashlib.sha256(extension.read_bytes()).hexdigest(), + })) tmp = tempfile.mkdtemp(prefix="duroxide-smoke-") db_path = os.path.join(tmp, "smoke.db") print( @@ -62,6 +73,16 @@ def hello_world(ctx, inp): finally: runtime.shutdown(200) + assert runtime.shutdown(0) is None + assert runtime.metrics_snapshot() is None + unstarted = Runtime(provider) + unstarted.shutdown(0) + try: + unstarted.start() + raise AssertionError("pre-start stop was not terminal") + except RuntimeError as error: + assert "lifecycle_terminal" in str(error) + return 0 diff --git a/docs/architecture.md b/docs/architecture.md index 8596ef4..247cb73 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -297,7 +297,82 @@ File-based SQLite can hit "database is locked" errors under concurrent orchestra ### Runtime Shutdown -`Runtime.shutdown(timeout_ms)` returns as soon as all runtime work has stopped. `timeout_ms` (default 1000) is the longest it waits for in-flight work before cancelling the rest; it is not a fixed delay. +In the unreleased lifecycle, `Runtime.shutdown(timeout_ms)` still returns `None` on success. +It waits for actual core retirement within supplied grace plus 5000 ms. +Omitted grace is 1000 ms. Idle execution can finish early. +Zero requests immediate force. Force does not prove that cleanup ended. +The first valid stop fixes both deadlines. +Repeated calls cannot obtain a new budget. + +The wrapper retains the prepared core Arc before it awaits fallible startup. +The first valid stop transfers that core and its callbacks to one independently owned native cleanup future. +The wrapper keeps the first deadline and completion record, not an obsolete callback registry. + +```mermaid +flowchart TD + Stop["First valid stop: fix grace and total"] --> Cleanup["Keep one native cleanup operation"] + Stop --> Wait["Wait within the original total"] + Cleanup -->|Cleanup proven| Release["Release obsolete core and activity-callback ownership"] + Cleanup -->|Cleanup unproven| Retain["Retain ownership"] + Wait -->|Completion published| Result["Return None or raise an operational error"] + Wait -->|Total expires first| Timeout["Raise TimeoutError; keep cleanup owned"] + Timeout --> Terminate["Application or supervisor must terminate the process"] +``` + +The caller's wait and cleanup have separate lifetimes. +Proven cleanup releases the core and its activity callbacks. +This includes quiescent failures and late completion without another shutdown call. +PyO3 defers off-GIL reference-count decrements until the next GIL entry. +A timeout does not cancel the cleanup owner. +Losing that owner or receiving an unproven result retains the core instead of releasing it. + +Upstream tracking includes native task destruction. +Provider futures can be canceled under the provider's atomic acknowledgment and recovery contract. +Repeated stop observes the retained completion or error. +It does not report success merely because the wrapper transferred its core. +Release does not clear the global orchestration-handler map or dispose independent Clients or providers. + +A valid pre-start stop is terminal. +Later registration and start calls fail. +Metrics remain unavailable after stop or accepted startup failure. +Serialize lifecycle and registration calls on the same instance. +This includes calls from other threads while the GIL is released. + +`TimeoutError` means that ordinary waiting ended while cleanup remains owned. +The application or supervisor must terminate the process, even after actual late completion. +Operational and startup failures raise `RuntimeError`; they do not report successful retirement. +The SDK never kills the process. +This adapter does not add the .NET foreign-continuation barrier or track arbitrary Python work. +Independent Clients, providers, and provider token-refresh work keep their separate lifetimes. +See the [lifecycle guide](user-guide.md#runtime-lifecycle) for validation and +release prerequisites. + +### Passive Stop Observation + +`Runtime` caches a receiver-only native source when it creates its wrapper. +`stop_observer()` uses this source without borrowing the mutable native Runtime +during startup or shutdown. Each observer owns metadata and a pending-wait slot, +not the Runtime, provider, or Python callback registry. + +Publisher retirement freezes the final state and detaches its live core receiver. +Core retirement alone is not SDK source end. A wait returns an unseen final +state before `None`; local `close()` instead makes pending and later use fail. +The final snapshot remains readable until local close. + +Foreground waiting releases the GIL. Native waits have bounded signal-check +opportunities, and Python signal handlers run outside metadata locks. +One absolute optional timeout is kept across those checks. +The pending-slot guard cancels/clears each return or error, including +`KeyboardInterrupt` and reentrant signal exceptions. +Native notification tasks own no Python or worker objects and cannot clear a +later wait's slot. Accepted oneshot-send counters are not Python-settlement counts. + +Notification final access is separate from public wait settlement and worker +cleanup. An observer can remain alive while eligible worker callbacks collect. +An offloaded Python wait still belongs to its calling thread: asyncio task +cancellation does not cancel it. A foreground/non-daemon waiting thread can +keep the process running. See the +[observer contract and example](user-guide.md#passive-stop-observation). ### Platform-Specific Binary diff --git a/docs/user-guide.md b/docs/user-guide.md index 921d91f..ae5ab2f 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -45,6 +45,188 @@ print(result.output) # "Hello, World!" runtime.shutdown() ``` +## Runtime Lifecycle + +This section requires the **unreleased** adapter and a matching core lifecycle build. +Current published packages and temporary local core overrides do not prove release readiness. + +Register handlers before the first `runtime.start()`. +Startup returns after activation, not after shutdown. Failures raise `RuntimeError`. +Registry and preparation failures become ordinary `RuntimeError` values with the `lifecycle_start_failed` category. +These include invalid or descending versions. +Subsequent shutdown calls retain those startup errors. +Version ordering and duplicate-registration policy do not change. +Adapter error messages exclude panic payloads. +This does not suppress Rust's separate process-wide panic-hook output. + +The normal `Runtime` constructor installs the SDK's generator drivers. +The underlying, unconfigured `PyRuntime` constructor is an internal setup boundary. +A missing-driver `lifecycle_start_failed` rejects before startup acceptance. +It therefore does not retain an activation failure. +Installing the real SDK drivers can still allow the first start. +Alternatively, a no-work shutdown succeeds and permanently prevents activation. +This distinction does not change retained failures after startup acceptance. +It does not add a supported low-level setup API. + +`runtime.shutdown()` uses one second of grace plus five seconds of total headroom. +`shutdown(0)` requests immediate force, with a five-second total. +The first stop fixes the deadlines. Repeated calls never receive a new budget. +Success remains `None`. A timeout raises `TimeoutError`. +Operational failures raise `RuntimeError`. + +Non-integer, negative, overflowing, or unsupported native durations fail validation. +Validation failure does not consume registrations or accept stop. +Unsupported or negative durations raise `ValueError`. +Non-integer extraction raises `TypeError`. +Values outside signed 64-bit extraction raise `OverflowError`. +The maximum grace interval is `((2**64 - 1) // 1_000_000) - 5000` milliseconds. +It must also fit the platform's monotonic clock range. +Validation runs before a no-work stop and before repeated calls. +`None` selects the omitted grace. + +Stopping before start is terminal. +Do not overlap lifecycle or registration calls on the same instance. +This includes calls from other threads while the GIL is released. +Metrics return `None` after accepted startup failure or stop. +Repeated shutdown can observe actual late completion, but a timeout permanently prevents reuse. +The application or supervisor must still terminate the process. +The SDK retains cleanup and never exits the process itself. + +This contract covers core-owned execution, not arbitrary unregistered Python work. +It does not add the .NET adapter's additional foreign-continuation barrier. +Independent provider-owned background work has a separate lifetime. +The adapter retains one native cleanup future independently of bounded caller waits. +Native Rust shutdown waits for cleanup without a total limit. +Provider futures can be canceled under the provider's atomic acknowledgment and peek-lock recovery contract. + +**Quiescence** means that tracked core execution and cleanup have ended. +Proven cleanup releases obsolete native activity-callback ownership. +This includes quiescent errors and actual late completion without another shutdown call. +The terminal result and original budgets remain available independently of those callbacks. +Unproven cleanup retains ownership. +Release does not clear the global orchestration-handler map or dispose independent providers or Clients. +It never removes a prior timeout's process-termination requirement. + +This complete example has no admitted work and can finish before grace: + +```python +from duroxide import Runtime, SqliteProvider + +provider = SqliteProvider.in_memory() +runtime = Runtime(provider) +runtime.start() +runtime.shutdown() # 1000 ms grace, 6000 ms total; can finish early +runtime.shutdown(0) # Observes the same actual completion, not a new stop +print(runtime.metrics_snapshot()) # None: stopped worker, not active metrics +``` + +A finite grace does not guarantee bounded forced cleanup. +Do not catch `TimeoutError` and continue using the worker or free foreign callback state. +Do not assume that a caught `RuntimeError` proves quiescence. +Report the failure to the application or supervisor responsible for process termination. + +The `lifecycle_shutdown_failed` category covers both a core failure after cleanup and loss of the shutdown owner without proven cleanup. +The category or exception type alone does not prove cleanup. +Owner loss still requires retained ownership and process termination. +At the first shutdown call, give the supervisor more than grace plus 5000 ms. +Leave additional time for other shutdown work and process termination. + +### Passive stop observation + +`runtime.stop_observer()` creates an independent `RuntimeStopObserver`. +Do not construct the observer directly. Creating it, reading `state`, and +waiting do not start or stop execution. +Observation can run during native startup or shutdown, including from another +thread. Lifecycle and registration calls themselves must still be serialized. + +| `RuntimeStopStatus` member | String value | Meaning | +|----------------------------|--------------|---------| +| `NOT_REQUESTED` | `"NotRequested"` | No stop is recorded. The worker can be unconstructed, starting, or running; this is not readiness or health. | +| `REQUESTED` | `"Requested"` | Stop was accepted, not necessarily received by native execution or completed. | +| `FAULTED` | `"Faulted"` | The core recorded a runtime-wide fault. It takes precedence and cannot regress. | + +Read the initial snapshot before waiting. A late observer can already be Faulted. +Invalid input, pre-core validation failure, wait timeout, Python interruption, +and observation errors do not create a runtime fault. + +This helper owns the observer passed to it. It reports the initial state and +each observed change, then ends its own observation after a fault. +Its finite per-wait timeout permits retry; it is not a total monitoring deadline. +Pass an observer rather than the Runtime so the waiting function need not +retain the worker: + +```python +from duroxide import RuntimeStopStatus + +def report_stop_changes(observer): + try: + state = observer.state + print("Runtime stop state:", state.value) + while state is not RuntimeStopStatus.FAULTED: + try: + next_state = observer.wait_for_change(state, timeout_ms=1000) + except TimeoutError: + print("No stop-state change before this timeout.") + continue + if next_state is None: + print("Stop observation source ended; cleanup is not proved.") + return + state = next_state + print("Runtime stop state:", state.value) + except KeyboardInterrupt: + print("Stop observation interrupted; worker shutdown is unchanged.") + except RuntimeError as error: + if not str(error).startswith("stop_observation_closed:"): + raise + print("Stop observation was closed.") + finally: + observer.close() +``` + +In an application that owns a Runtime, obtain `observer = runtime.stop_observer()` +and call `report_stop_changes(observer)` where blocking is acceptable. +Keep the observer reference if another thread must close it. +The application's lifecycle owner must still stop the Runtime normally. +Unexpected observation failures propagate; do not treat them as a recorded +runtime fault or successful cleanup. + +`wait_for_change` compares against the supplied last-seen state; it is not an +event queue. Changes can coalesce. A final unseen state arrives before `None`. +The final snapshot stays readable until local close. +The source can remain open after worker cleanup. An unchanged REQUESTED or +FAULTED therefore still waits; close monitoring when you no longer need it. + +Only one wait can be pending on an observer; overlap raises `RuntimeError` with +`stop_observation_busy`. Use separate observers for independent readers. +`timeout_ms=None` means no timeout. A nonnegative integer sets a timeout in +milliseconds and must fit the native deadline range. +Zero returns an available change or normal source end immediately; otherwise +it raises `TimeoutError`. Timeout and interruption consume no state and permit +retry. The helper above closes its observer when it exits. + +`close()` is idempotent and can run from another thread while a wait is active. +Pending and later operations raise `RuntimeError` with `stop_observation_closed`. +Close does not rewrite a result that already returned. +It does not request shutdown or join cleanup or native notification final access. + +Native waiting releases the GIL and checks pending Python signals at bounded +opportunities. A `KeyboardInterrupt` or an exception from a signal handler +propagates from the wait. Signal handling and notification delivery still need +GIL/executor progress; these are not real-time scheduling guarantees. +The optional timeout uses one deadline; signal checks do not restart it. + +The API is synchronous. Do not block an asyncio event-loop thread with it. +Canceling a task that uses `asyncio.to_thread` does not stop that thread's +underlying wait. Use a finite wait or close the observer explicitly when ending +the offloaded monitor. A foreground or non-daemon Python thread can still keep +the process running. + +The observer and native notification tasks retain metadata, not Runtime, +provider, or callback registry ownership. Neither state nor source end proves +quiescence or permits release. This API does not add ownership of arbitrary +Python work, renew first-stop budgets, remove a termination requirement, or +select host/process policy. + ## Orchestration Patterns ### Sequential Steps diff --git a/python/duroxide/__init__.py b/python/duroxide/__init__.py index c5fe570..cf484a5 100644 --- a/python/duroxide/__init__.py +++ b/python/duroxide/__init__.py @@ -57,6 +57,7 @@ PyRuntimeOptions = RuntimeOptions import json +from enum import Enum # ─── Generator Driver Registry ─────────────────────────────────── @@ -420,6 +421,81 @@ def prune_executions_bulk(self, filter=None, options=None): return self._native.prune_executions_bulk(filter, options) +class RuntimeStopStatus(str, Enum): + """String-valued stop metadata, not readiness or proof of cleanup. + + NOT_REQUESTED includes unconstructed, starting and running workers. + REQUESTED records accepted stop, not native receipt or completed cleanup. + FAULTED is a sticky recorded core-wide fault. Observation errors, timeout, + interruption and pre-core validation failures do not produce it. + """ + + NOT_REQUESTED = "NotRequested" + REQUESTED = "Requested" + FAULTED = "Faulted" + + +_STOP_OBSERVER_TOKEN = object() + + +class RuntimeStopObserver: + """Owned, metadata-only observation from Runtime.stop_observer(). + + Read state before waiting: a late reader can already see FAULTED. + Each consumer needs an independent observer and owns its close(). + Observation retains no Runtime/provider/registry ownership, requests no + shutdown, and changes no shutdown deadline or application policy. + Neither state nor normal source end proves cleanup or permits release. + """ + + def __init__(self, native_observer=None, *, _token=None): + if _token is not _STOP_OBSERVER_TOKEN: + raise TypeError("Use Runtime.stop_observer() to create an observer") + self._native = native_observer + + @property + def state(self) -> RuntimeStopStatus: + """Read retained state without starting or stopping execution. + + The final snapshot remains readable after normal source end, until + local close. Access after close raises RuntimeError. + """ + return RuntimeStopStatus(self._native.state) + + def wait_for_change(self, last_seen: RuntimeStopStatus, timeout_ms: int = None): + """Wait for a state different from last_seen, or None at normal source end. + + Return an unseen final state before None. Intermediate changes can + coalesce. Unchanged REQUESTED/FAULTED still waits while the source is open. + Only one wait may be pending on this observer; overlap raises RuntimeError. + + Use timeout_ms=None for no timeout, or a nonnegative integer number of + milliseconds in the native deadline range. Zero returns an available + change/source end immediately, otherwise raises TimeoutError. + Expiry and KeyboardInterrupt permit retry and consume no state. + Local close instead raises RuntimeError and prevents reuse. + Observation errors propagate without manufacturing FAULTED. + + Native waiting releases the GIL and checks Python signals at bounded + opportunities outside metadata locks. Signal-handler exceptions propagate. + One absolute timeout is kept; signal checks do not restart it. + The calling thread still blocks and delivery needs GIL/executor progress. + Do not block an asyncio event loop. Canceling an asyncio.to_thread task + does not cancel the underlying wait; use close() or a finite timeout. + """ + status = self._native.wait_for_change(last_seen, timeout_ms) + return RuntimeStopStatus(status) if status is not None else None + + def close(self): + """End this observer, not worker cleanup. Idempotent and terminal. + + Safe from another thread while a wait is active. Pending and later + operations raise RuntimeError; an already returned result is unchanged. + Does not join worker cleanup or native notification final access. + """ + self._native.close() + + class Runtime: """Durable execution runtime.""" @@ -428,6 +504,7 @@ def __init__(self, provider, options=None): self._native = PyRuntime.from_postgres(provider._native, options) else: self._native = PyRuntime.from_sqlite(provider._native, options) + self._stop_source = self._native.stop_source() # Wire up the generator driver functions self._native.set_generator_driver( create_generator, next_step, dispose_generator @@ -505,13 +582,13 @@ def my_orch(ctx, input): runtime.register_orchestration("my_orch", my_fn) """ if fn is not None: - _orchestration_functions[name] = fn self._native.register_orchestration(name) + _orchestration_functions[name] = fn return fn def decorator(func): - _orchestration_functions[name] = func self._native.register_orchestration(name) + _orchestration_functions[name] = func return func return decorator @@ -526,14 +603,14 @@ def register_orchestration_versioned(self, name: str, version: str, fn=None): """Register a versioned orchestration generator function. Can be used as a decorator.""" if fn is not None: key = f"{name}@{version}" - _orchestration_functions[key] = fn self._native.register_orchestration_versioned(name, version) + _orchestration_functions[key] = fn return fn def decorator(func): key = f"{name}@{version}" - _orchestration_functions[key] = func self._native.register_orchestration_versioned(name, version) + _orchestration_functions[key] = func return func return decorator @@ -548,16 +625,40 @@ def my_orch(ctx, input): """ return self.register_orchestration_versioned(name, version, fn) + def stop_observer(self) -> RuntimeStopObserver: + """Create an independent observer, including during native startup/shutdown. + + Does not start/stop execution or retain this worker. Read its initial + state before waiting. The caller owns close(). + """ + return RuntimeStopObserver( + self._stop_source.observer(), _token=_STOP_OBSERVER_TOKEN + ) + def start(self): - """Start the runtime. Blocks until shutdown is called.""" + """Start once and return after startup. + + Do not overlap lifecycle or registration calls on this instance. + """ self._native.start() def shutdown(self, timeout_ms: int = None): - """Shutdown the runtime gracefully.""" + """Stop permanently, even before start. + + Grace defaults to 1000 ms. Total waiting is grace plus 5000 ms. + Repeated calls keep the original deadlines and observe actual completion or errors. + TimeoutError leaves cleanup owned. + The application or supervisor must then terminate the process. + Startup and operational failures raise RuntimeError. + RuntimeError alone does not prove that cleanup ended. + Loss of the native shutdown owner also requires process termination because cleanup is unproven. + Durations must be nonnegative integers in the native deadline range. + Do not overlap lifecycle or registration calls on this instance. + """ self._native.shutdown(timeout_ms) def metrics_snapshot(self): - """Get a snapshot of runtime metrics.""" + """Get runtime metrics, or None after accepted startup failure or stop.""" return self._native.metrics_snapshot() @@ -600,6 +701,8 @@ def default_and(tags: list) -> str: "PostgresEntraOptions", "Client", "Runtime", + "RuntimeStopStatus", + "RuntimeStopObserver", "OrchestrationResult", "RuntimeOptions", "OrchestrationContext", diff --git a/src/lib.rs b/src/lib.rs index b8ed197..e2c066a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,9 @@ mod handlers; mod pg_provider; mod provider; mod runtime; +mod stop_observation; +#[cfg(feature = "test-hooks")] +mod test_hooks; mod types; use pyo3::prelude::*; @@ -173,6 +176,11 @@ fn init_tracing( #[pymodule] fn _duroxide(m: &Bound<'_, PyModule>) -> PyResult<()> { + #[cfg(feature = "test-hooks")] + { + m.add_function(wrap_pyfunction!(test_hooks::_lifecycle_test_hooks, m)?)?; + m.add_class::()?; + } m.add_function(wrap_pyfunction!(activity_trace_log, m)?)?; m.add_function(wrap_pyfunction!(orchestration_trace_log, m)?)?; m.add_function(wrap_pyfunction!(activity_is_cancelled, m)?)?; @@ -195,6 +203,8 @@ fn _duroxide(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/src/runtime.rs b/src/runtime.rs index 9699cdc..4af26dd 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -1,9 +1,12 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. +use pyo3::exceptions::{PyRuntimeError, PyTimeoutError, PyValueError}; use pyo3::prelude::*; +use std::panic::{catch_unwind, AssertUnwindSafe}; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; +use tokio::sync::watch; use duroxide::providers::TagFilter; use duroxide::runtime::{self, OrchestrationHandler, OrchestrationRegistry}; @@ -11,6 +14,7 @@ use duroxide::runtime::{self, OrchestrationHandler, OrchestrationRegistry}; use crate::handlers::{PyActivityHandler, PyOrchestrationHandler}; use crate::pg_provider::PyPostgresProvider; use crate::provider::PySqliteProvider; +use crate::stop_observation::{PyRuntimeStopSource, StopSource}; use crate::types::PyMetricsSnapshot; /// Global tokio runtime shared by all Python-facing blocking methods. @@ -152,6 +156,13 @@ pub struct PyRuntime { dispose_fn: Option>, options: Option, inner: Option>, + started: bool, + shutdown_requested: bool, + stop: Option, + startup_error: Option, + pub(crate) stop_source: StopSource, + #[cfg(feature = "test-hooks")] + pub(crate) test_hooks: runtime::test_hooks::LifecycleHooks, } #[pymethods] @@ -169,6 +180,13 @@ impl PyRuntime { dispose_fn: None, options, inner: None, + started: false, + shutdown_requested: false, + stop: None, + startup_error: None, + stop_source: StopSource::new(), + #[cfg(feature = "test-hooks")] + test_hooks: runtime::test_hooks::LifecycleHooks::default(), } } @@ -185,9 +203,20 @@ impl PyRuntime { dispose_fn: None, options, inner: None, + started: false, + shutdown_requested: false, + stop: None, + startup_error: None, + stop_source: StopSource::new(), + #[cfg(feature = "test-hooks")] + test_hooks: runtime::test_hooks::LifecycleHooks::default(), } } + fn stop_source(&self) -> PyRuntimeStopSource { + self.stop_source.subscribe() + } + /// Set the generator driver functions (called once from Python before registering orchestrations). /// These three functions handle: creating generators, driving next steps, and disposing. fn set_generator_driver( @@ -195,195 +224,394 @@ impl PyRuntime { create_fn: Py, next_fn: Py, dispose_fn: Py, - ) { + ) -> PyResult<()> { + self.ensure_configurable()?; self.create_fn = Some(create_fn); self.next_fn = Some(next_fn); self.dispose_fn = Some(dispose_fn); + Ok(()) } /// Register a Python activity function. /// The function receives a payload string and returns a result string. - fn register_activity(&mut self, name: String, callback: Py) { + fn register_activity(&mut self, name: String, callback: Py) -> PyResult<()> { + self.ensure_configurable()?; self.activity_builders.push((name, callback)); + Ok(()) } /// Register a Python orchestration (generator function). - fn register_orchestration(&mut self, name: String) { + fn register_orchestration(&mut self, name: String) -> PyResult<()> { + self.ensure_configurable()?; self.orchestration_names.push((name, None)); + Ok(()) } /// Register a versioned Python orchestration. - fn register_orchestration_versioned(&mut self, name: String, version: String) { + fn register_orchestration_versioned(&mut self, name: String, version: String) -> PyResult<()> { + self.ensure_configurable()?; self.orchestration_names.push((name, Some(version))); + Ok(()) } /// Start the runtime. This processes orchestrations and activities until shutdown. fn start(&mut self, py: Python<'_>) -> PyResult<()> { - // Build activity registry - let mut activity_builder = duroxide::runtime::registry::ActivityRegistry::builder(); - for (name, callback) in self.activity_builders.drain(..) { - let handler = Arc::new(PyActivityHandler::new(name.clone(), callback)); - activity_builder = activity_builder.register(&name, move |ctx, input| { - let h = handler.clone(); - async move { h.invoke(ctx, input).await } - }); + self.ensure_configurable()?; + if self.create_fn.is_none() || self.next_fn.is_none() || self.dispose_fn.is_none() { + return Err(PyRuntimeError::new_err( + "lifecycle_start_failed: generator driver not set", + )); } - let activities = activity_builder.build(); + self.started = true; + let prepared = catch_unwind(AssertUnwindSafe(|| { + // Build activity registry + let mut activity_builder = duroxide::runtime::registry::ActivityRegistry::builder(); + for (name, callback) in self.activity_builders.drain(..) { + let handler = Arc::new(PyActivityHandler::new(name.clone(), callback)); + activity_builder = activity_builder.register(&name, move |ctx, input| { + let h = handler.clone(); + async move { h.invoke(ctx, input).await } + }); + } + let activities = activity_builder.build(); - // Build orchestration registry - let create_fn = self.create_fn.take().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err( - "Generator driver not set. Call set_generator_driver() before start().", - ) - })?; - let next_fn = self.next_fn.take().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Generator driver not set.") - })?; - let dispose_fn = self.dispose_fn.take().ok_or_else(|| { - pyo3::exceptions::PyRuntimeError::new_err("Generator driver not set.") - })?; + // Build orchestration registry + let create_fn = self + .create_fn + .take() + .ok_or(runtime::RuntimeStartError::StartupFailed)?; + let next_fn = self + .next_fn + .take() + .ok_or(runtime::RuntimeStartError::StartupFailed)?; + let dispose_fn = self + .dispose_fn + .take() + .ok_or(runtime::RuntimeStartError::StartupFailed)?; - let mut orch_builder = OrchestrationRegistry::builder(); - for (name, version) in self.orchestration_names.drain(..) { - let handler = Arc::new(PyOrchestrationHandler::new( - create_fn.clone_ref(py), - next_fn.clone_ref(py), - dispose_fn.clone_ref(py), - )); - if let Some(ver) = version { - orch_builder = - orch_builder.register_versioned(&name, &ver, move |ctx, input| { + let mut orch_builder = OrchestrationRegistry::builder(); + for (name, version) in self.orchestration_names.drain(..) { + let handler = Arc::new(PyOrchestrationHandler::new( + create_fn.clone_ref(py), + next_fn.clone_ref(py), + dispose_fn.clone_ref(py), + )); + if let Some(ver) = version { + orch_builder = + orch_builder.register_versioned(&name, &ver, move |ctx, input| { + let h = handler.clone(); + async move { h.invoke(ctx, input).await } + }); + } else { + orch_builder = orch_builder.register(&name, move |ctx, input| { let h = handler.clone(); async move { h.invoke(ctx, input).await } }); - } else { - orch_builder = orch_builder.register(&name, move |ctx, input| { - let h = handler.clone(); - async move { h.invoke(ctx, input).await } - }); + } } - } - let orchestrations = orch_builder.build(); + let orchestrations = orch_builder.build(); - // Build runtime options - let mut rt_options = runtime::RuntimeOptions::default(); - if let Some(ref opts) = self.options { - if let Some(c) = opts.orchestration_concurrency { - rt_options.orchestration_concurrency = c as usize; - } - if let Some(c) = opts.worker_concurrency { - rt_options.worker_concurrency = c as usize; - } - if let Some(ms) = opts.dispatcher_poll_interval_ms { - rt_options.dispatcher_min_poll_interval = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.worker_lock_timeout_ms { - rt_options.worker_lock_timeout = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.worker_lock_renewal_buffer_ms { - rt_options.worker_lock_renewal_buffer = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.orchestrator_lock_timeout_ms { - rt_options.orchestrator_lock_timeout = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.orchestrator_lock_renewal_buffer_ms { - rt_options.orchestrator_lock_renewal_buffer = Duration::from_millis(ms as u64); - } - if let Some(n) = opts.max_attempts { - rt_options.max_attempts = n as u32; - } - if let Some(ref fmt) = opts.log_format { - rt_options.observability.log_format = match fmt.as_str() { - "json" => runtime::LogFormat::Json, - "pretty" => runtime::LogFormat::Pretty, - _ => runtime::LogFormat::Compact, - }; - } - if let Some(ref level) = opts.log_level { - rt_options.observability.log_level = level.clone(); - } - if let Some(ref name) = opts.service_name { - rt_options.observability.service_name = name.clone(); - } - if let Some(ref ver) = opts.service_version { - rt_options.observability.service_version = Some(ver.clone()); - } - if let Some(max) = opts.max_sessions_per_runtime { - rt_options.max_sessions_per_runtime = max as usize; - } - if let Some(ms) = opts.session_idle_timeout_ms { - rt_options.session_idle_timeout = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.session_lock_timeout_ms { - rt_options.session_lock_timeout = Duration::from_millis(ms as u64); - } - if let Some(ms) = opts.session_lock_renewal_buffer_ms { - rt_options.session_lock_renewal_buffer = Duration::from_millis(ms as u64); - } - if let Some(ref nid) = opts.worker_node_id { - rt_options.worker_node_id = Some(nid.clone()); + // Build runtime options + let mut rt_options = runtime::RuntimeOptions::default(); + if let Some(ref opts) = self.options { + if let Some(c) = opts.orchestration_concurrency { + rt_options.orchestration_concurrency = c as usize; + } + if let Some(c) = opts.worker_concurrency { + rt_options.worker_concurrency = c as usize; + } + if let Some(ms) = opts.dispatcher_poll_interval_ms { + rt_options.dispatcher_min_poll_interval = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.worker_lock_timeout_ms { + rt_options.worker_lock_timeout = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.worker_lock_renewal_buffer_ms { + rt_options.worker_lock_renewal_buffer = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.orchestrator_lock_timeout_ms { + rt_options.orchestrator_lock_timeout = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.orchestrator_lock_renewal_buffer_ms { + rt_options.orchestrator_lock_renewal_buffer = Duration::from_millis(ms as u64); + } + if let Some(n) = opts.max_attempts { + rt_options.max_attempts = n as u32; + } + if let Some(ref fmt) = opts.log_format { + rt_options.observability.log_format = match fmt.as_str() { + "json" => runtime::LogFormat::Json, + "pretty" => runtime::LogFormat::Pretty, + _ => runtime::LogFormat::Compact, + }; + } + if let Some(ref level) = opts.log_level { + rt_options.observability.log_level = level.clone(); + } + if let Some(ref name) = opts.service_name { + rt_options.observability.service_name = name.clone(); + } + if let Some(ref ver) = opts.service_version { + rt_options.observability.service_version = Some(ver.clone()); + } + if let Some(max) = opts.max_sessions_per_runtime { + rt_options.max_sessions_per_runtime = max as usize; + } + if let Some(ms) = opts.session_idle_timeout_ms { + rt_options.session_idle_timeout = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.session_lock_timeout_ms { + rt_options.session_lock_timeout = Duration::from_millis(ms as u64); + } + if let Some(ms) = opts.session_lock_renewal_buffer_ms { + rt_options.session_lock_renewal_buffer = Duration::from_millis(ms as u64); + } + if let Some(ref nid) = opts.worker_node_id { + rt_options.worker_node_id = Some(nid.clone()); + } + if let Some(ref filter_str) = opts.worker_tag_filter { + rt_options.worker_tag_filter = parse_tag_filter(filter_str); + } } - if let Some(ref filter_str) = opts.worker_tag_filter { - rt_options.worker_tag_filter = parse_tag_filter(filter_str); + + // Release GIL before blocking — orchestration handlers need GIL access + let provider = self.provider.clone(); + #[cfg(feature = "test-hooks")] + let provider = Arc::new(runtime::test_hooks::TestProvider::new( + provider, + self.test_hooks.clone(), + )); + runtime::Runtime::prepare(provider, activities, orchestrations, rt_options) + })); + let rt = match prepared { + Ok(result) => result, + Err(payload) => { + discard_panic_payload(payload); + Err(runtime::RuntimeStartError::StartupFailed) } } - - // Release GIL before blocking — orchestration handlers need GIL access - let provider = self.provider.clone(); - let rt = py.allow_threads(|| { - TOKIO_RT.block_on(async { - runtime::Runtime::start_with_options( - provider, - activities, - orchestrations, - rt_options, - ) - .await + .map_err(|error| { + self.startup_error = Some(error); + start_error(error) + })?; + self.stop_source.attach(rt.stop_observer()); + self.inner = Some(rt.clone()); + #[cfg(feature = "test-hooks")] + rt.set_lifecycle_hooks(self.test_hooks.clone()) + .map_err(|error| { + self.startup_error = Some(error); + start_error(error) + })?; + py.allow_threads(|| TOKIO_RT.block_on(rt.start_execution())) + .map_err(|error| { + self.startup_error = Some(error); + start_error(error) }) - }); - - self.inner = Some(rt); - Ok(()) } /// Get a snapshot of runtime metrics. fn metrics_snapshot(&self) -> Option { - self.inner.as_ref()?.metrics_snapshot().map(|m| PyMetricsSnapshot { - orch_starts: m.orch_starts, - orch_completions: m.orch_completions, - orch_failures: m.orch_failures, - orch_application_errors: m.orch_application_errors, - orch_infrastructure_errors: m.orch_infrastructure_errors, - orch_configuration_errors: m.orch_configuration_errors, - orch_poison: m.orch_poison, - activity_success: m.activity_success, - activity_app_errors: m.activity_app_errors, - activity_infra_errors: m.activity_infra_errors, - activity_config_errors: m.activity_config_errors, - activity_poison: m.activity_poison, - orch_dispatcher_items_fetched: m.orch_dispatcher_items_fetched, - worker_dispatcher_items_fetched: m.worker_dispatcher_items_fetched, - orch_continue_as_new: m.orch_continue_as_new, - suborchestration_calls: m.suborchestration_calls, - provider_errors: m.provider_errors, - }) + if self.shutdown_requested || self.startup_error.is_some() { + return None; + } + self.inner + .as_ref()? + .metrics_snapshot() + .map(|m| PyMetricsSnapshot { + orch_starts: m.orch_starts, + orch_completions: m.orch_completions, + orch_failures: m.orch_failures, + orch_application_errors: m.orch_application_errors, + orch_infrastructure_errors: m.orch_infrastructure_errors, + orch_configuration_errors: m.orch_configuration_errors, + orch_poison: m.orch_poison, + activity_success: m.activity_success, + activity_app_errors: m.activity_app_errors, + activity_infra_errors: m.activity_infra_errors, + activity_config_errors: m.activity_config_errors, + activity_poison: m.activity_poison, + orch_dispatcher_items_fetched: m.orch_dispatcher_items_fetched, + worker_dispatcher_items_fetched: m.worker_dispatcher_items_fetched, + orch_continue_as_new: m.orch_continue_as_new, + suborchestration_calls: m.suborchestration_calls, + provider_errors: m.provider_errors, + }) } /// Shutdown the runtime gracefully. #[pyo3(signature = (timeout_ms=None))] fn shutdown(&mut self, py: Python<'_>, timeout_ms: Option) -> PyResult<()> { - if let Some(rt) = self.inner.take() { - let timeout = timeout_ms.map(|ms| ms as u64); - py.allow_threads(|| { - TOKIO_RT.block_on(async { - rt.shutdown(timeout).await; + let (grace_deadline, total_deadline) = shutdown_deadlines(timeout_ms)?; + self.stop_source.requested(); + if self.stop.is_none() { + if let Some(runtime) = &self.inner { + runtime.request_shutdown(); + let (completed, completion) = watch::channel(None); + let owner = ShutdownOwner { + runtime: self.inner.take(), + }; + self.stop = Some(ShutdownOperation { + completion, + total_deadline, }); - }); + TOKIO_RT.spawn(async move { + completed.send_replace(Some(owner.shutdown_until(grace_deadline).await)); + }); + } + } + self.shutdown_requested = true; + if let Some(stop) = self.stop.clone() { + py.allow_threads(|| TOKIO_RT.block_on(stop.wait())) + .map_err(shutdown_error)?; + } else { + self.stop_source.close(); + self.activity_builders.clear(); + self.orchestration_names.clear(); + self.create_fn = None; + self.next_fn = None; + self.dispose_fn = None; + } + if let Some(error) = self.startup_error { + return Err(start_error(error)); + } + Ok(()) + } +} + +impl PyRuntime { + fn ensure_configurable(&self) -> PyResult<()> { + if self.started || self.shutdown_requested { + return Err(PyRuntimeError::new_err( + "lifecycle_terminal: runtime has already started or shutdown was requested", + )); } Ok(()) } } +pub(crate) fn discard_panic_payload(payload: Box) { + // Payload destruction must not reintroduce a panic across the language boundary. + if let Err(secondary) = catch_unwind(AssertUnwindSafe(|| drop(payload))) { + std::mem::forget(secondary); + } +} + +fn start_error(error: runtime::RuntimeStartError) -> PyErr { + PyRuntimeError::new_err(format!("lifecycle_start_failed: {error}")) +} + +struct ShutdownOwner { + runtime: Option>, +} + +impl ShutdownOwner { + async fn shutdown_until( + mut self, + grace_deadline: Instant, + ) -> Result { + let result = self + .runtime + .as_ref() + .unwrap() + .clone() + .shutdown_until(grace_deadline) + .await; + let quiescent = match &result { + Ok(_) => true, + Err(error) => error.is_quiescent(), + }; + if quiescent { + // PyO3 defers off-GIL reference decrements until the next GIL entry. + self.runtime = None; + } + result + } +} + +impl Drop for ShutdownOwner { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + // Losing the owner, including an unpolled/aborted task, is not + // cleanup proof. Retain the callback-bearing core in that case. + std::mem::forget(runtime); + } + } +} + +#[derive(Clone)] +struct ShutdownOperation { + completion: + watch::Receiver>>, + total_deadline: Instant, +} + +#[derive(Clone, Copy)] +enum ShutdownWaitError { + TimedOut, + OwnerLost, + Core(runtime::RuntimeShutdownError), +} + +impl ShutdownOperation { + async fn wait(mut self) -> Result { + loop { + { + let completed = self.completion.borrow_and_update(); + if let Some(result) = *completed { + return result.map_err(ShutdownWaitError::Core); + } + if Instant::now() >= self.total_deadline { + return Err(ShutdownWaitError::TimedOut); + } + } + tokio::select! { + biased; + changed = self.completion.changed() => { + if changed.is_err() { + return Err(ShutdownWaitError::OwnerLost); + } + }, + () = tokio::time::sleep_until(self.total_deadline.into()) => {}, + } + } + } +} + +fn shutdown_error(error: ShutdownWaitError) -> PyErr { + match error { + ShutdownWaitError::TimedOut => PyTimeoutError::new_err( + "lifecycle_shutdown_timed_out: cleanup remains owned and incomplete; application or supervisor must terminate the process", + ), + ShutdownWaitError::OwnerLost => PyRuntimeError::new_err( + "lifecycle_shutdown_failed: native shutdown owner failed; quiescence is unproven; application or supervisor must terminate the process", + ), + ShutdownWaitError::Core(error) => PyRuntimeError::new_err(format!("lifecycle_shutdown_failed: {error}")), + } +} + +fn shutdown_deadlines(timeout_ms: Option) -> PyResult<(Instant, Instant)> { + let invalid = || { + PyValueError::new_err( + "lifecycle_invalid_timeout: timeout_ms must be a nonnegative supported millisecond duration", + ) + }; + let millis = u64::try_from(timeout_ms.unwrap_or(1000)).map_err(|_| invalid())?; + let grace = Duration::from_millis(millis); + let total = grace + .checked_add(Duration::from_secs(5)) + .ok_or_else(invalid)?; + if total > Duration::from_nanos(u64::MAX) { + return Err(invalid()); + } + let now = Instant::now(); + let grace_deadline = now.checked_add(grace).ok_or_else(invalid)?; + let total_deadline = now.checked_add(total).ok_or_else(invalid)?; + total_deadline + .checked_add(Duration::from_millis(1)) + .ok_or_else(invalid)?; + Ok((grace_deadline, total_deadline)) +} + /// Parse a tag filter string into a TagFilter. /// /// Accepts: diff --git a/src/stop_observation.rs b/src/stop_observation.rs new file mode 100644 index 0000000..120f2ff --- /dev/null +++ b/src/stop_observation.rs @@ -0,0 +1,383 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +use duroxide::runtime::{RuntimeStopObserver, RuntimeStopStatus}; +use parking_lot::Mutex; +use pyo3::exceptions::{PyRuntimeError, PyTimeoutError, PyValueError}; +use pyo3::prelude::*; +use std::time::{Duration, Instant}; +use tokio::sync::{oneshot, watch}; + +use crate::runtime::TOKIO_RT; + +const SIGNAL_CHECK_INTERVAL: Duration = Duration::from_millis(50); + +#[derive(Clone)] +struct SourceState { + status: RuntimeStopStatus, + core: Option, + closed: bool, +} + +fn merge(left: RuntimeStopStatus, right: RuntimeStopStatus) -> RuntimeStopStatus { + use RuntimeStopStatus::*; + match (left, right) { + (Faulted, _) | (_, Faulted) => Faulted, + (Requested, _) | (_, Requested) => Requested, + _ => NotRequested, + } +} + +impl SourceState { + fn status(&self) -> RuntimeStopStatus { + merge( + self.status, + self.core.as_ref().map_or(self.status, |core| core.state()), + ) + } +} + +pub(crate) struct StopSource { + state: watch::Sender, + #[cfg(feature = "test-hooks")] + pub(crate) hooks: crate::test_hooks::ObservationHooks, +} + +impl StopSource { + pub fn new() -> Self { + Self { + state: watch::channel(SourceState { + status: RuntimeStopStatus::NotRequested, + core: None, + closed: false, + }) + .0, + #[cfg(feature = "test-hooks")] + hooks: crate::test_hooks::ObservationHooks::default(), + } + } + + pub fn subscribe(&self) -> PyRuntimeStopSource { + PyRuntimeStopSource { + state: self.state.subscribe(), + #[cfg(feature = "test-hooks")] + hooks: self.hooks.clone(), + } + } + + pub fn attach(&self, core: RuntimeStopObserver) { + self.state.send_modify(|state| { + assert!(!state.closed && state.core.is_none()); + state.status = merge(state.status, core.state()); + state.core = Some(core); + }); + } + + pub fn requested(&self) { + self.state.send_if_modified(|state| { + let status = merge(state.status, RuntimeStopStatus::Requested); + if state.closed || state.status == status { + return false; + } + state.status = status; + true + }); + } + + pub fn close(&self) { + self.state.send_modify(|state| { + state.status = state.status(); + state.core = None; + state.closed = true; + }); + } +} + +impl Drop for StopSource { + fn drop(&mut self) { + // Freeze the final snapshot instead of leaving a live core receiver behind. + self.close(); + } +} + +#[pyclass] +pub struct PyRuntimeStopSource { + state: watch::Receiver, + #[cfg(feature = "test-hooks")] + hooks: crate::test_hooks::ObservationHooks, +} + +#[pymethods] +impl PyRuntimeStopSource { + fn observer(&self) -> PyRuntimeStopObserver { + PyRuntimeStopObserver { + state: Mutex::new(ObserverState { + source: Some(self.state.clone()), + pending: None, + }), + #[cfg(feature = "test-hooks")] + hooks: self.hooks.clone(), + } + } +} + +struct ObserverState { + source: Option>, + pending: Option>, +} + +#[pyclass] +pub struct PyRuntimeStopObserver { + state: Mutex, + #[cfg(feature = "test-hooks")] + hooks: crate::test_hooks::ObservationHooks, +} + +#[pymethods] +impl PyRuntimeStopObserver { + #[getter] + fn state(&self) -> PyResult<&'static str> { + let state = self.state.lock(); + let source = state.source.as_ref().ok_or_else(closed_error)?; + let snapshot = source.borrow(); + Ok(encode(snapshot.status())) + } + + #[pyo3(signature = (last_seen, timeout_ms=None))] + fn wait_for_change( + &self, + py: Python<'_>, + last_seen: &str, + timeout_ms: Option, + ) -> PyResult> { + let last_seen = decode(last_seen)?; + let deadline = observation_deadline(timeout_ms)?; + let (source, mut canceled) = { + let mut state = self.state.lock(); + let source = state.source.as_ref().ok_or_else(closed_error)?; + if state.pending.is_some() { + return Err(PyRuntimeError::new_err( + "stop_observation_busy: this observer already has a pending wait", + )); + } + { + let snapshot = source.borrow(); + let status = snapshot.status(); + if status != last_seen { + return Ok(Some(encode(status))); + } + if snapshot.closed { + return Ok(None); + } + } + if deadline.is_some_and(|until| Instant::now() >= until) { + return Err(timeout_error()); + } + let source = source.clone(); + let (cancel, canceled) = watch::channel(()); + state.pending = Some(cancel); + (source, canceled) + }; + let _pending = PendingWait(&self.state); + let mut completion = spawn_wait( + source, + last_seen, + canceled.clone(), + #[cfg(feature = "test-hooks")] + self.hooks.clone(), + ); + loop { + let check_signals_at = Instant::now() + SIGNAL_CHECK_INTERVAL; + let wake_at = deadline.map_or(check_signals_at, |until| until.min(check_signals_at)); + let result = py.allow_threads(|| { + TOKIO_RT.block_on(async { + tokio::select! { + biased; + _ = canceled.changed() => Some(Err(ObservationError::Closed)), + result = &mut completion => Some(result.unwrap_or(Err(ObservationError::Failed))), + () = tokio::time::sleep_until(wake_at.into()) => None, + } + }) + }); + // Signal handlers may reenter close/state/wait. Never run them under a metadata lock. + py.check_signals()?; + if self.state.lock().source.is_none() { + return Err(closed_error()); + } + let result = result.or_else(|| match completion.try_recv() { + Ok(result) => Some(result), + Err(oneshot::error::TryRecvError::Empty) => None, + Err(oneshot::error::TryRecvError::Closed) => Some(Err(ObservationError::Failed)), + }); + if let Some(result) = result { + return result + .map(|status| status.map(encode)) + .map_err(|error| match error { + ObservationError::Closed => closed_error(), + ObservationError::Failed => PyRuntimeError::new_err( + "stop_observation_failed: native observation failed", + ), + }); + } + if deadline.is_some_and(|until| Instant::now() >= until) { + return Err(timeout_error()); + } + } + } + + fn close(&self) { + let mut state = self.state.lock(); + state.source = None; + state.pending = None; + } +} + +struct PendingWait<'a>(&'a Mutex); + +impl Drop for PendingWait<'_> { + fn drop(&mut self) { + self.0.lock().pending = None; + } +} + +enum ObservationError { + Closed, + Failed, +} + +fn spawn_wait( + mut source: watch::Receiver, + last_seen: RuntimeStopStatus, + mut canceled: watch::Receiver<()>, + #[cfg(feature = "test-hooks")] hooks: crate::test_hooks::ObservationHooks, +) -> oneshot::Receiver, ObservationError>> { + let (completed, completion) = oneshot::channel(); + #[cfg(feature = "test-hooks")] + let pending = hooks.register(); + #[cfg(feature = "test-hooks")] + let waiting_hooks = hooks.clone(); + // Both tasks own metadata only. No Python reference or runtime owner crosses this boundary. + let waiting = TOKIO_RT.spawn(async move { + tokio::select! { + biased; + _ = canceled.changed() => Err(ObservationError::Closed), + status = async { + #[cfg(feature = "test-hooks")] + waiting_hooks.checkpoint("observer-wait").await; + changed(&mut source, last_seen).await + } => Ok(status), + } + }); + TOKIO_RT.spawn(async move { + let result = match waiting.await { + Ok(result) => result, + Err(error) => { + if error.is_panic() { + crate::runtime::discard_panic_payload(error.into_panic()); + } + Err(ObservationError::Failed) + } + }; + #[cfg(feature = "test-hooks")] + let result = if hooks.checkpoint("observer-delivery").await { + Err(ObservationError::Failed) + } else { + result + }; + // The caller can already have timed out, been interrupted or closed. + let delivered = completed.send(result).is_ok(); + #[cfg(feature = "test-hooks")] + { + hooks.delivered(delivered); + hooks.checkpoint("observer-final-access").await; + drop(pending); + } + #[cfg(not(feature = "test-hooks"))] + let _ = delivered; + }); + completion +} + +async fn changed( + source: &mut watch::Receiver, + last_seen: RuntimeStopStatus, +) -> Option { + loop { + let (state, status) = { + let state = source.borrow_and_update(); + (state.clone(), state.status()) + }; + if status != last_seen { + return Some(status); + } + if state.closed { + return None; + } + if let Some(mut core) = state.core { + let native_status = core.state(); + if merge(state.status, native_status) != last_seen { + continue; + } + tokio::select! { + _ = source.changed() => {}, + result = core.changed(native_status) => { + if result.is_none() { + // Core cleanup does not end a live SDK publisher. + let _ = source.changed().await; + } + }, + } + } else { + let _ = source.changed().await; + } + } +} + +fn encode(status: RuntimeStopStatus) -> &'static str { + match status { + RuntimeStopStatus::NotRequested => "NotRequested", + RuntimeStopStatus::Requested => "Requested", + RuntimeStopStatus::Faulted => "Faulted", + } +} + +fn decode(status: &str) -> PyResult { + match status { + "NotRequested" => Ok(RuntimeStopStatus::NotRequested), + "Requested" => Ok(RuntimeStopStatus::Requested), + "Faulted" => Ok(RuntimeStopStatus::Faulted), + _ => Err(PyValueError::new_err( + "stop_observation_invalid_status: invalid last-seen status", + )), + } +} + +fn observation_deadline(timeout_ms: Option) -> PyResult> { + let Some(millis) = timeout_ms else { + return Ok(None); + }; + let invalid = || { + PyValueError::new_err( + "stop_observation_invalid_timeout: timeout_ms must be a nonnegative supported millisecond duration", + ) + }; + let duration = Duration::from_millis(u64::try_from(millis).map_err(|_| invalid())?); + if duration > Duration::from_nanos(u64::MAX) { + return Err(invalid()); + } + let deadline = Instant::now().checked_add(duration).ok_or_else(invalid)?; + deadline + .checked_add(Duration::from_millis(1)) + .ok_or_else(invalid)?; + Ok(Some(deadline)) +} + +fn closed_error() -> PyErr { + PyRuntimeError::new_err("stop_observation_closed: this observer is closed") +} + +fn timeout_error() -> PyErr { + PyTimeoutError::new_err( + "stop_observation_timed_out: no changed state was delivered before the deadline", + ) +} diff --git a/src/test_hooks.rs b/src/test_hooks.rs new file mode 100644 index 0000000..f114a65 --- /dev/null +++ b/src/test_hooks.rs @@ -0,0 +1,237 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +use duroxide::runtime::test_hooks::{ + LifecycleHooks, LifecyclePoint, LifecycleTask, ProviderOperation, +}; +use pyo3::exceptions::{PyRuntimeError, PyTimeoutError}; +use pyo3::prelude::*; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::watch; + +use crate::runtime::{PyRuntime, TOKIO_RT}; +use crate::stop_observation::StopSource; + +#[pyclass] +pub struct LifecycleTestHooks { + hooks: LifecycleHooks, + observation: ObservationHooks, +} + +#[pyfunction] +pub fn _lifecycle_test_hooks(runtime: &PyRuntime) -> LifecycleTestHooks { + LifecycleTestHooks { + hooks: runtime.test_hooks.clone(), + observation: runtime.stop_source.hooks.clone(), + } +} + +#[pymethods] +impl LifecycleTestHooks { + fn hold(&self, point: &str) -> PyResult<()> { + if point.starts_with("observer-") { + return self.observation.hold(point); + } + let parsed = parse_point(point)?; + let gate = if point == "blocking-provider" { + self.hooks.hold_blocking(parsed) + } else { + self.hooks.hold(parsed) + }; + gate.map(|_| ()) + .map_err(|error| PyRuntimeError::new_err(format!("test gate: {error:?}"))) + } + + fn release(&self, point: &str) -> PyResult<()> { + if point.starts_with("observer-") { + self.observation.gate(point)?.released.send_replace(true); + return Ok(()); + } + self.hooks + .gate(parse_point(point)?) + .ok_or_else(|| PyRuntimeError::new_err("test gate not armed"))? + .release(); + Ok(()) + } + + fn wait_entered(&self, py: Python<'_>, point: &str) -> PyResult<()> { + if point.starts_with("observer-") { + let gate = self.observation.gate(point)?; + return py + .allow_threads(|| { + TOKIO_RT.block_on(async { + tokio::time::timeout( + Duration::from_secs(3), + signaled(gate.entered.subscribe()), + ) + .await + }) + }) + .map_err(|_| PyTimeoutError::new_err("test gate entry timed out")); + } + let gate = self + .hooks + .gate(parse_point(point)?) + .ok_or_else(|| PyRuntimeError::new_err("test gate not armed"))?; + py.allow_threads(|| { + TOKIO_RT.block_on(async { + tokio::time::timeout(Duration::from_secs(3), gate.entered()).await + }) + }) + .map_err(|_| PyTimeoutError::new_err("test gate entry timed out")) + } + + fn fail(&self, point: &str, message: &str) -> PyResult<()> { + if point == "observer-delivery" { + self.observation.0.lock().failures.insert(point.to_owned()); + return Ok(()); + } + self.hooks + .fail_once_with_message(parse_point(point)?, message) + .map_err(|error| PyRuntimeError::new_err(format!("test fault: {error:?}"))) + } + + fn retire_observation_publisher(&self, runtime: &mut PyRuntime) -> PyResult<()> { + if !Arc::ptr_eq(&self.observation.0, &runtime.stop_source.hooks.0) { + return Err(PyRuntimeError::new_err( + "test observation publisher does not match", + )); + } + drop(std::mem::replace( + &mut runtime.stop_source, + StopSource::new(), + )); + Ok(()) + } + + fn snapshot(&self) -> String { + let counts = self.hooks.counts(); + let observation = self.observation.0.lock(); + serde_json::json!({ + "started": counts.iter().map(|count| count.started).sum::(), + "completed": counts.iter().map(|count| count.completed).sum::(), + "active": counts.iter().map(|count| count.active).sum::(), + "cleanupActive": self.hooks.cleanup_counts().active, + "coordinators": self.hooks.hits(LifecyclePoint::ShutdownCoordinator), + "forceRequests": self.hooks.hits(LifecyclePoint::ForceRequested), + "observerPending": observation.pending, + "observerDeliveries": observation.deliveries, + }) + .to_string() + } +} + +#[derive(Clone, Default)] +pub(crate) struct ObservationHooks(Arc>); + +#[derive(Default)] +struct ObservationState { + gates: HashMap>, + failures: HashSet, + pending: usize, + deliveries: usize, +} + +struct ObservationGate { + entered: watch::Sender, + released: watch::Sender, +} + +async fn signaled(mut signal: watch::Receiver) { + while !*signal.borrow_and_update() { + signal + .changed() + .await + .expect("test gate sender is retained"); + } +} + +impl ObservationHooks { + fn hold(&self, point: &str) -> PyResult<()> { + if !matches!( + point, + "observer-wait" | "observer-delivery" | "observer-final-access" + ) { + return Err(PyRuntimeError::new_err("unknown observation test point")); + } + let mut state = self.0.lock(); + if state + .gates + .get(point) + .is_some_and(|gate| !*gate.released.borrow()) + { + return Err(PyRuntimeError::new_err( + "observation test gate already armed", + )); + } + state.gates.insert( + point.to_owned(), + Arc::new(ObservationGate { + entered: watch::channel(false).0, + released: watch::channel(false).0, + }), + ); + Ok(()) + } + + fn gate(&self, point: &str) -> PyResult> { + self.0 + .lock() + .gates + .get(point) + .cloned() + .ok_or_else(|| PyRuntimeError::new_err("observation test gate not armed")) + } + + pub(crate) async fn checkpoint(&self, point: &str) -> bool { + let (gate, failed) = { + let mut state = self.0.lock(); + ( + state.gates.get(point).cloned(), + state.failures.remove(point), + ) + }; + if let Some(gate) = gate { + gate.entered.send_replace(true); + signaled(gate.released.subscribe()).await; + } + failed + } + + pub(crate) fn register(&self) -> ObservationTask { + self.0.lock().pending += 1; + ObservationTask(self.clone()) + } + + pub(crate) fn delivered(&self, delivered: bool) { + if delivered { + self.0.lock().deliveries += 1; + } + } +} + +pub(crate) struct ObservationTask(ObservationHooks); + +impl Drop for ObservationTask { + fn drop(&mut self) { + self.0 .0.lock().pending -= 1; + } +} + +fn parse_point(point: &str) -> PyResult { + match point { + "startup" => Ok(LifecyclePoint::StartupBeforeIo), + "partial-startup" => Ok(LifecyclePoint::StartupAfterSpawn( + LifecycleTask::OrchestrationSlot, + )), + "provider" | "blocking-provider" => Ok(LifecyclePoint::ProviderReturn( + ProviderOperation::FetchOrchestration, + )), + "worker" => Ok(LifecyclePoint::ParentWork(LifecycleTask::WorkerSlot)), + "force" => Ok(LifecyclePoint::ForceRequested), + "coordinator" => Ok(LifecyclePoint::ShutdownCoordinator), + _ => Err(PyRuntimeError::new_err("unknown lifecycle test point")), + } +} diff --git a/tests/fixtures/lifecycle_callbacks.py b/tests/fixtures/lifecycle_callbacks.py new file mode 100644 index 0000000..0a96a40 --- /dev/null +++ b/tests/fixtures/lifecycle_callbacks.py @@ -0,0 +1,128 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import gc +import json +import time +import weakref + +from duroxide import Client, Runtime, RuntimeOptions, SqliteProvider +from duroxide import _duroxide as native + + +class ActivityOwner: + def __init__(self, runtime): + self.runtime = runtime + + def activity(self, _context, value): + return value + + +def attach_activity_owner(runtime, capture): + owner = ActivityOwner(runtime) + if capture == "bound": + callback = owner.activity + elif capture == "self-capturing": + callback = lambda context, value: owner.activity(context, value) + else: + callback = lambda _context, value: value + runtime.register_activity("CallbackCollection-" + capture, callback) + return { + "owner": weakref.ref(owner), + "runtime": weakref.ref(runtime), + "callback": weakref.ref(callback), + } + + +def collection_state(references): + return {name: reference() is None for name, reference in references.items()} + + +def settle_deferred_references(references, deadline=None): + if deadline is None: + deadline = time.perf_counter_ns() + 2_000_000_000 + while True: + # A fresh PyO3 entry flushes queued off-GIL decrefs; task counts alone + # would not show whether an opaque callback cycle was actually released. + native.RuntimeOptions() + gc.collect() + collected = collection_state(references) + if all(collected.values()) or time.perf_counter_ns() >= deadline: + return collected + time.sleep(0.001) + + +def stopped_owner(capture, scenario, database_directory): + # Canceled SQLx acquisitions can discard the last in-memory connection. + # A file store isolates provider ownership from that connection-lifetime rule. + provider = SqliteProvider.open(str(database_directory / f"{capture}.db")) + client = Client(provider) + runtime = Runtime(provider, RuntimeOptions( + dispatcher_poll_interval_ms=5, + session_idle_timeout_ms=0 if scenario == "callbacks-preparation-failed" else None, + )) + references = attach_activity_owner(runtime, capture) + hooks = None + start_error = None + stop_error = None + if scenario == "callbacks-operational-failed": + hooks = native._lifecycle_test_hooks(runtime._native) + hooks.hold("worker") + hooks.fail("worker", "callback-collection-fault") + if scenario != "callbacks-prestart": + try: + runtime.start() + assert scenario != "callbacks-preparation-failed" + except RuntimeError as error: + assert scenario == "callbacks-preparation-failed" + assert "lifecycle_start_failed" in str(error) + start_error = str(error) + if hooks is not None: + try: + hooks.wait_entered("worker") + finally: + hooks.release("worker") + for timeout in (100, 30_000): + try: + assert runtime.shutdown(timeout) is None + assert scenario not in ("callbacks-preparation-failed", "callbacks-operational-failed") + except RuntimeError as error: + if scenario == "callbacks-operational-failed": + assert "lifecycle_shutdown_failed" in str(error) + assert "cleanup completed and is quiescent" in str(error) + assert "terminate the process" not in str(error) + else: + assert scenario == "callbacks-preparation-failed" + assert str(error) == start_error + if stop_error is not None: + assert str(error) == stop_error + stop_error = str(error) + if hooks is not None: + state = json.loads(hooks.snapshot()) + assert state["active"] == state["cleanupActive"] == 0 + assert state["started"] == state["completed"] + assert runtime.metrics_snapshot() is None + for operation in ( + runtime.start, + lambda: runtime.register_activity("CallbackCollection-late", lambda _context, value: value), + lambda: runtime.register_orchestration("CallbackCollection-late", lambda _context, value: value), + ): + try: + operation() + raise AssertionError("quiescent runtime was reusable") + except RuntimeError as error: + assert "lifecycle_terminal" in str(error) + return references, provider, client, stop_error + + +def collect_stopped_owners(scenario, database_directory): + cases = {} + for capture in ("noncapturing", "bound", "self-capturing"): + # The creating frame must be gone before inspecting any weak reference. + references, provider, client, stop_error = stopped_owner(capture, scenario, database_directory) + collected = settle_deferred_references(references) + # Independent clients/providers must still work after callback release. + client.start_orchestration("after-callback-cleanup", "Unregistered", None) + assert provider is not None + cases[capture] = {"collected": collected, "stopError": stop_error} + return cases diff --git a/tests/fixtures/lifecycle_observers.py b/tests/fixtures/lifecycle_observers.py new file mode 100644 index 0000000..86f60d7 --- /dev/null +++ b/tests/fixtures/lifecycle_observers.py @@ -0,0 +1,510 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import _thread +import gc +import json +from pathlib import Path +import signal +import sys +import threading +import time +import traceback +import weakref + +from duroxide import Client, Runtime, RuntimeOptions, RuntimeStopStatus, SqliteProvider +from duroxide import _duroxide as native + +from lifecycle_callbacks import attach_activity_owner, settle_deferred_references + + +NOT_REQUESTED = RuntimeStopStatus.NOT_REQUESTED +REQUESTED = RuntimeStopStatus.REQUESTED +FAULTED = RuntimeStopStatus.FAULTED + + +def fixture(database=None): + provider = SqliteProvider.open(str(database)) if database else SqliteProvider.in_memory() + runtime = Runtime(provider, RuntimeOptions(dispatcher_poll_interval_ms=5)) + hooks = native._lifecycle_test_hooks(runtime._native) if hasattr(native, "_lifecycle_test_hooks") else None + return runtime, provider, hooks + + +def expect_error(error_type, operation, category=None): + try: + operation() + except error_type as error: + assert type(error) is error_type, type(error) + if category is not None: + assert category in str(error), str(error) + return str(error) + raise AssertionError(f"expected {error_type.__name__}") + + +class Call: + def __init__(self, function, *args): + self.results = [] + self.thread = threading.Thread(target=self._run, args=(function, args), daemon=True) + self.thread.start() + + def _run(self, function, args): + try: + self.results.append(("returned", function(*args))) + except BaseException as error: + # Do not keep exception tracebacks that can retain a creating worker frame. + self.results.append(("raised", type(error).__name__, str(error))) + + def outcome(self, timeout=3): + self.thread.join(timeout) + assert not self.thread.is_alive(), "observer probe call did not finish" + assert len(self.results) == 1, self.results + return self.results[0] + + +def until_busy(observer, last_seen): + deadline = time.perf_counter() + 2 + while True: + try: + observer.wait_for_change(last_seen, 0) + except TimeoutError: + pass + except RuntimeError as error: + assert "stop_observation_busy" in str(error), str(error) + return + else: + raise AssertionError("wait was not pending on the expected state") + assert time.perf_counter() < deadline, "native wait did not register" + time.sleep(0.001) + + +def until_no_native_waits(hooks): + deadline = time.perf_counter() + 2 + while json.loads(hooks.snapshot())["observerPending"]: + assert time.perf_counter() < deadline, hooks.snapshot() + time.sleep(0.001) + + +def assert_raised(call, error_type, category): + outcome = call.outcome() + assert outcome[:2] == ("raised", error_type.__name__), outcome + assert category in outcome[2], outcome + return outcome + + +def active_worker(): + runtime, provider, hooks = fixture() + client = Client(provider) + entered, release, completed = threading.Event(), threading.Event(), threading.Event() + + @runtime.register_activity("ObserverHeldCallback") + def callback(_context, value): + entered.set() + release.wait() + completed.set() + return value + + @runtime.register_orchestration("ObserverHeldFlow") + def flow(context, value): + return (yield context.schedule_activity("ObserverHeldCallback", value)) + + runtime.start() + client.start_orchestration("observer-held", "ObserverHeldFlow", "observed") + assert entered.wait(3), "foreign callback did not enter" + return runtime, client, hooks, release, completed + + +def interruption(action): + runtime, client, hooks, release, completed = active_worker() + observer, peer = runtime.stop_observer(), runtime.stop_observer() + progress, controller_errors, signal_calls = [], [], [] + previous_handler = signal.getsignal(signal.SIGINT) + + def handler(_number, _frame): + assert observer.state is NOT_REQUESTED + expect_error(RuntimeError, lambda: observer.wait_for_change(NOT_REQUESTED, 0), "stop_observation_busy") + signal_calls.append("reentered") + if action == "signal-close": + observer.close() + elif action == "signal-error": + raise ValueError("observer signal handler failed") + + if action.startswith("signal-"): + signal.signal(signal.SIGINT, handler) + + def control(): + try: + until_busy(observer, NOT_REQUESTED) + assert observer.state is NOT_REQUESTED + expect_error(TimeoutError, lambda: peer.wait_for_change(NOT_REQUESTED, 0)) + progress.append("native-wait-active") + if action == "close": + observer.close() + elif action != "timeout": + _thread.interrupt_main() + except BaseException: + controller_errors.append(traceback.format_exc()) + observer.close() + + controller = threading.Thread(target=control, daemon=True) + controller.start() + before = time.perf_counter() + try: + if action in ("timeout", "signal-return"): + expect_error(TimeoutError, lambda: observer.wait_for_change(NOT_REQUESTED, 175)) + elif action == "interrupt": + expect_error(KeyboardInterrupt, lambda: observer.wait_for_change(NOT_REQUESTED)) + elif action == "signal-error": + expect_error(ValueError, lambda: observer.wait_for_change(NOT_REQUESTED), "observer signal handler failed") + else: + expect_error(RuntimeError, lambda: observer.wait_for_change(NOT_REQUESTED), "stop_observation_closed") + elapsed = time.perf_counter() - before + controller.join(2) + assert not controller.is_alive() + assert not controller_errors, controller_errors + assert progress == ["native-wait-active"], progress + assert not release.is_set() and not completed.is_set() + assert runtime.metrics_snapshot() is not None + assert peer.state is NOT_REQUESTED + if hooks: + state = json.loads(hooks.snapshot()) + assert state["active"] > 0 and state["coordinators"] == 0, state + if action in ("close", "signal-close"): + observer.close() + expect_error(RuntimeError, lambda: observer.state, "stop_observation_closed") + expect_error(RuntimeError, lambda: observer.wait_for_change(NOT_REQUESTED), "stop_observation_closed") + else: + assert observer.state is NOT_REQUESTED + expect_error(TimeoutError, lambda: observer.wait_for_change(NOT_REQUESTED, 0)) + if action in ("timeout", "signal-return"): + assert 0.15 <= elapsed < 1, elapsed + else: + assert elapsed < 1, elapsed + if action.startswith("signal-"): + assert signal_calls == ["reentered"], signal_calls + return { + "action": action, "elapsedSeconds": elapsed, + "otherThreadProgress": True, "nativeWaitActive": True, + "callbackHeldAtSettlement": True, "stopStateAtSettlement": peer.state, + "signalHandlerReentered": bool(signal_calls), + } + finally: + signal.signal(signal.SIGINT, previous_handler) + observer.close() + peer.close() + release.set() + assert client.wait_for_orchestration("observer-held", 3000).status == "Completed" + runtime.shutdown(100) + + +def ordered_races(): + outcomes = {"change": 0, "close": 0, "timeout": 0, "precore": 0} + for iteration in range(100): + runtime, _, hooks = fixture() + observer, peer = runtime.stop_observer(), runtime.stop_observer() + action = ("change", "close", "timeout", "precore")[iteration % 4] + if hooks: + hooks.hold("observer-wait") + first = Call(observer.wait_for_change, NOT_REQUESTED, 175 if action == "timeout" else None) + second = Call(peer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + until_busy(peer, NOT_REQUESTED) + if hooks: + hooks.wait_entered("observer-wait") + assert observer.state is NOT_REQUESTED + if action != "precore": + runtime.start() + assert observer.state is NOT_REQUESTED + until_busy(peer, NOT_REQUESTED) + if action == "close": + observer.close() + assert_raised(first, RuntimeError, "stop_observation_closed") + until_busy(peer, NOT_REQUESTED) + elif action == "timeout": + assert_raised(first, TimeoutError, "stop_observation_timed_out") + expect_error(TimeoutError, lambda: observer.wait_for_change(NOT_REQUESTED, 0)) + first = Call(observer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + until_busy(peer, NOT_REQUESTED) + runtime.shutdown(100) + if hooks: + hooks.release("observer-wait") + assert second.outcome() == ("returned", REQUESTED), second.results + if action != "close": + assert first.outcome() == ("returned", REQUESTED), first.results + assert observer.state is REQUESTED + assert peer.state is REQUESTED + if action == "precore": + assert peer.wait_for_change(REQUESTED, 0) is None + else: + expect_error(TimeoutError, lambda: peer.wait_for_change(REQUESTED, 0)) + observer.close() + peer.close() + if hooks: + until_no_native_waits(hooks) + counts = json.loads(hooks.snapshot()) + assert counts["active"] == counts["cleanupActive"] == 0, counts + assert counts["coordinators"] == (0 if action == "precore" else 1), counts + assert len(first.results) == len(second.results) == 1 + outcomes[action] += 1 + return { + "iterations": 100, "outcomes": outcomes, "duplicateSettlements": 0, + "nativeRegistrationHeld": hooks is not None, + } + + +def attachment_faults(): + for _ in range(100): + runtime, _, hooks = fixture() + hooks.hold("startup") + hooks.hold("worker") + hooks.hold("blocking-provider") + hooks.fail("worker", "observer-private-sentinel") + observer = runtime.stop_observer() + before_attach = Call(observer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + startup = Call(runtime.start) + hooks.wait_entered("startup") + during_start = runtime.stop_observer() + assert during_start.state is NOT_REQUESTED + expect_error(TimeoutError, lambda: during_start.wait_for_change(NOT_REQUESTED, 0)) + hooks.release("startup") + assert startup.outcome() == ("returned", None), startup.results + hooks.wait_entered("worker") + hooks.wait_entered("blocking-provider") + stopping = Call(runtime.shutdown, 30_000) + assert before_attach.outcome() == ("returned", REQUESTED), before_attach.results + during_stop = runtime.stop_observer() + assert during_stop.state is REQUESTED + fault_wait = Call(observer.wait_for_change, REQUESTED) + until_busy(observer, REQUESTED) + hooks.release("worker") + assert fault_wait.outcome() == ("returned", FAULTED), fault_wait.results + assert during_start.wait_for_change(NOT_REQUESTED, 0) is FAULTED + assert during_stop.wait_for_change(REQUESTED, 0) is FAULTED + late = runtime.stop_observer() + assert late.state is FAULTED + assert stopping.thread.is_alive() + assert json.loads(hooks.snapshot())["active"] > 0 + observer.close() + assert late.state is FAULTED + hooks.release("blocking-provider") + error = assert_raised(stopping, RuntimeError, "cleanup completed and is quiescent") + assert "sentinel" not in error[2] and "terminate the process" not in error[2] + assert late.state is FAULTED + expect_error(TimeoutError, lambda: late.wait_for_change(FAULTED, 0)) + for item in (during_start, during_stop, late): + item.close() + until_no_native_waits(hooks) + counts = json.loads(hooks.snapshot()) + assert counts["active"] == counts["cleanupActive"] == 0, counts + # Core fault rollback can join the SDK's stop, as in the existing lifecycle controls. + assert 1 <= counts["coordinators"] <= 2, counts + return {"iterations": 100, "faultObservedBeforeCleanup": True, "duplicateSettlements": 0} + + +def fault_before_stop(): + runtime, client, hooks, release, completed = active_worker() + observer, late = runtime.stop_observer(), None + try: + waiting = Call(observer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + hooks.fail("provider", "observer-foreign-private-sentinel") + client.start_orchestration("observer-fault-trigger", "ObserverHeldFlow", "trigger") + assert waiting.outcome() == ("returned", FAULTED), waiting.results + assert observer.state is FAULTED + assert not completed.is_set(), "foreign callback retired before fault observation" + counts = json.loads(hooks.snapshot()) + assert counts["active"] + counts["cleanupActive"] > 0, counts + late = runtime.stop_observer() + assert late.state is FAULTED + assert late.wait_for_change(NOT_REQUESTED, 0) is FAULTED + expect_error(TimeoutError, lambda: late.wait_for_change(FAULTED, 0)) + assert not completed.is_set() + finally: + observer.close() + if late is not None: + late.close() + release.set() + error = expect_error(RuntimeError, lambda: runtime.shutdown(100), "cleanup completed and is quiescent") + assert "sentinel" not in error + assert completed.wait(1) + until_no_native_waits(hooks) + return { + "faultWithoutSdkStop": True, "callbackHeldAtFault": True, + "lateState": FAULTED, "callbackRetiredAfterRelease": True, + } + + +def source_freeze(): + runtime, _, hooks = fixture() + for point in ("worker", "blocking-provider", "coordinator"): + hooks.hold(point) + hooks.fail("worker", "observer-source-private-sentinel") + runtime.start() + hooks.wait_entered("worker") + hooks.wait_entered("blocking-provider") + observer = runtime.stop_observer() + waiting = Call(observer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + hooks.retire_observation_publisher(runtime._native) + assert waiting.outcome() == ("returned", None), waiting.results + assert observer.state is NOT_REQUESTED + hooks.release("worker") + hooks.wait_entered("coordinator") + assert json.loads(hooks.snapshot())["active"] > 0 + assert observer.state is NOT_REQUESTED + assert observer.wait_for_change(NOT_REQUESTED, 0) is None + hooks.release("blocking-provider") + hooks.release("coordinator") + expect_error(RuntimeError, lambda: runtime.shutdown(100), "cleanup completed and is quiescent") + assert observer.state is NOT_REQUESTED + assert observer.wait_for_change(NOT_REQUESTED, 0) is None + observer.close() + return {"sourceEndedBeforeFault": True, "finalState": NOT_REQUESTED} + + +def source_end(): + def unstarted(): + runtime, _, _ = fixture() + return runtime.stop_observer(), weakref.ref(runtime) + + observer, runtime_reference = unstarted() + gc.collect() + assert runtime_reference() is None + assert observer.state is NOT_REQUESTED + assert observer.wait_for_change(NOT_REQUESTED, 1000) is None + observer.close() + + def stopped(): + runtime, _, _ = fixture() + references = attach_activity_owner(runtime, "self-capturing") + observer, peer = runtime.stop_observer(), runtime.stop_observer() + runtime.start() + runtime.shutdown(100) + return observer, peer, references + + observer, peer, references = stopped() + collected = settle_deferred_references(references) + assert all(collected.values()), collected + assert observer.state is REQUESTED + assert observer.wait_for_change(NOT_REQUESTED, 0) is REQUESTED + assert observer.wait_for_change(REQUESTED, 0) is None + observer.close() + assert peer.state is REQUESTED + assert peer.wait_for_change(NOT_REQUESTED, 0) is REQUESTED + assert peer.wait_for_change(REQUESTED, 0) is None + peer.close() + return {"collected": collected, "unstartedOwnerCollected": True, "unseenFinalState": REQUESTED} + + +def collection(retain_delivery=False): + runtime, provider, hooks = fixture(Path(sys.argv[2]) / "collection.db") + references = attach_activity_owner(runtime, "bound") + worker_references = {name: reference for name, reference in references.items() if name != "runtime"} + finalized = [] + for name, reference in worker_references.items(): + weakref.finalize(reference(), finalized.append, name) + observer = runtime.stop_observer() + runtime.start() + if retain_delivery: + hooks.hold("observer-delivery") + hooks.hold("observer-final-access") + first = Call(observer.wait_for_change, NOT_REQUESTED, 175 if retain_delivery else None) + until_busy(observer, NOT_REQUESTED) + runtime.shutdown(100) + if retain_delivery: + hooks.wait_entered("observer-delivery") + assert_raised(first, TimeoutError, "stop_observation_timed_out") + else: + assert first.outcome() == ("returned", REQUESTED), first.results + retry = Call(observer.wait_for_change, REQUESTED) + until_busy(observer, REQUESTED) + if retain_delivery: + hooks.release("observer-delivery") + hooks.wait_entered("observer-final-access") + assert json.loads(hooks.snapshot())["observerPending"] == 2 + until_busy(observer, REQUESTED) + collected = settle_deferred_references(worker_references) + assert all(collected.values()), collected + assert set(finalized) == set(worker_references), finalized + assert references["runtime"]() is runtime + assert observer.state is REQUESTED + assert retry.thread.is_alive() + Client(provider).start_orchestration("independent-after-observation", "Unregistered", None) + if hooks: + state = json.loads(hooks.snapshot()) + assert state["active"] == state["cleanupActive"] == 0, state + assert state["started"] == state["completed"], state + assert state["observerPending"] >= 1, state + observer.close() + assert_raised(retry, RuntimeError, "stop_observation_closed") + if retain_delivery: + assert json.loads(hooks.snapshot())["observerPending"] == 2 + hooks.release("observer-final-access") + if hooks: + until_no_native_waits(hooks) + return { + "collected": collected, "finalized": sorted(finalized), + "runtimeRetained": True, "observerRetained": True, + "sourceOpenDuringCollection": True, "nativeTailHeld": retain_delivery, + "oldDeliveryDidNotSettleRetry": retain_delivery, + } + + +def transport_failure(): + runtime, _, hooks = fixture() + observer, peer = runtime.stop_observer(), runtime.stop_observer() + hooks.fail("observer-delivery", "observation-private-sentinel") + waiting = Call(observer.wait_for_change, NOT_REQUESTED) + until_busy(observer, NOT_REQUESTED) + runtime.shutdown(0) + error = assert_raised(waiting, RuntimeError, "stop_observation_failed") + assert "sentinel" not in error[2] + assert observer.state is peer.state is REQUESTED + assert observer.wait_for_change(NOT_REQUESTED, 0) is REQUESTED + assert observer.wait_for_change(REQUESTED, 0) is None + observer.close() + peer.close() + until_no_native_waits(hooks) + return {"failureSanitized": True, "stateAfterFailure": REQUESTED, "retrySucceeded": True} + + +def natural_exit(): + runtime, _, hooks = fixture() + observer = runtime.stop_observer() + runtime.start() + runtime.shutdown(100) + if hooks: + hooks.hold("observer-final-access") + expect_error(TimeoutError, lambda: observer.wait_for_change(REQUESTED, 20)) + if hooks: + hooks.wait_entered("observer-final-access") + counts = json.loads(hooks.snapshot()) + assert counts["active"] == counts["cleanupActive"] == 0 + assert counts["observerPending"] == 1 + globals()["retained_runtime"] = runtime + globals()["retained_observer"] = observer + globals()["retained_hooks"] = hooks + return {"runtimeRetained": True, "observerRetained": True, "nativeTailHeld": hooks is not None} + + +def run_observer_scenario(scenario, emit): + actions = ("timeout", "close", "interrupt", "signal-close", "signal-error", "signal-return") + action = scenario.removeprefix("observer-") + if action in actions: + result = interruption(action) + else: + result = { + "ordered-races": ordered_races, + "attachment-faults": attachment_faults, + "source-freeze": source_freeze, + "fault-before-stop": fault_before_stop, + "source-end": source_end, + "collection": collection, + "delivery-retention": lambda: collection(True), + "transport-failure": transport_failure, + "exit": natural_exit, + }[action]() + emit("OBSERVER_RESULT", scenario=scenario, **result) + assert sys.stdin.readline().strip() == "finish" + emit("OBSERVER_FINISHED") diff --git a/tests/fixtures/lifecycle_probe.py b/tests/fixtures/lifecycle_probe.py new file mode 100644 index 0000000..aa1783c --- /dev/null +++ b/tests/fixtures/lifecycle_probe.py @@ -0,0 +1,167 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import gc +import json +from pathlib import Path +import sys +import threading +import time +import traceback + +from duroxide import Runtime, RuntimeOptions, SqliteProvider +from duroxide import _duroxide as native + +from lifecycle_callbacks import ( + attach_activity_owner, + collect_stopped_owners, + collection_state, + settle_deferred_references, +) +from lifecycle_observers import run_observer_scenario + + +def emit(phase, **detail): + clock = time.get_clock_info("perf_counter") + print("LIFECYCLE_PROBE " + json.dumps({ + "phase": phase, + "clock": {"implementation": clock.implementation, "resolutionSeconds": clock.resolution}, + **detail, + }), flush=True) + + +def main(): + scenario = sys.argv[1] + if scenario.startswith("observer-"): + run_observer_scenario(scenario, emit) + return + if scenario in ( + "callbacks-prestart", "callbacks-stopped", "callbacks-preparation-failed", + "callbacks-operational-failed", + ): + emit("COLLECTION", cases=collect_stopped_owners(scenario, Path(sys.argv[2]))) + assert sys.stdin.readline().strip() == "finish" + emit("COLLECTION_FINISHED") + return + runtime = Runtime(SqliteProvider.in_memory(), RuntimeOptions( + dispatcher_poll_interval_ms=5, service_name="lifecycle-private-sentinel", + )) + if scenario in ("registry-invalid", "registry-descending"): + versions = ["not-semver"] if scenario == "registry-invalid" else ["2.0.0", "1.0.0"] + for version in versions: + runtime.register_orchestration_versioned( + "registry-private-sentinel", version, lambda _context, _input: "ok", + ) + try: + runtime.start() + raise AssertionError("invalid registry unexpectedly started") + except RuntimeError as error: + assert type(error) is RuntimeError + start_message = str(error) + assert "lifecycle_start_failed" in start_message + assert "sentinel" not in start_message + for timeout in (0, 30_000): + try: + runtime.shutdown(timeout) + raise AssertionError("failed startup was replaced by successful shutdown") + except RuntimeError as error: + assert str(error) == start_message + assert runtime.metrics_snapshot() is None + for operation in ( + runtime.start, + lambda: runtime.register_activity("late", lambda _context, _input: "late"), + lambda: runtime.register_orchestration("late", lambda _context, _input: "late"), + lambda: runtime.register_orchestration_versioned("late", "3.0.0", lambda _context, _input: "late"), + ): + try: + operation() + raise AssertionError("failed startup was reusable") + except RuntimeError as error: + assert "lifecycle_terminal" in str(error) + globals()["retained_runtime"] = runtime + emit("REGISTRY_FAILURE_RETAINED", scenario=scenario) + return + if scenario == "exit": + runtime.register_activity("ExitActivity", lambda _context, value: value) + runtime.register_orchestration("ExitFlow", lambda _context, _input: "ok") + runtime.start() + runtime.shutdown(100) + globals()["retained_runtime"] = runtime + emit("EXIT_READY") + return + references = attach_activity_owner(runtime, "bound") + hooks = native._lifecycle_test_hooks(runtime._native) + hooks.hold("blocking-provider") + runtime.start() + hooks.wait_entered("blocking-provider") + emit("STOPPING") + start = time.perf_counter_ns() + try: + runtime.shutdown(None if scenario == "late" else 0) + raise AssertionError("held provider falsely completed") + except TimeoutError as error: + assert "lifecycle_shutdown_timed_out" in str(error) + assert "cleanup remains owned and incomplete" in str(error) + assert "terminate the process" in str(error) + assert "sentinel" not in str(error) + elapsed = (time.perf_counter_ns() - start) / 1_000_000_000 + repeated = time.perf_counter_ns() + try: + runtime.shutdown(30_000) + raise AssertionError("repeated stop falsely completed") + except TimeoutError as error: + assert "lifecycle_shutdown_timed_out" in str(error) + assert "sentinel" not in str(error) + repeated = (time.perf_counter_ns() - repeated) / 1_000_000_000 + assert json.loads(hooks.snapshot())["active"] > 0 + assert runtime.metrics_snapshot() is None + assert not any(collection_state(references).values()) + emit("TIMED_OUT", elapsedSeconds=elapsed, repeatedSeconds=repeated) + if sys.stdin.readline().strip() == "retain": + runtime = None + gc.collect() + state = json.loads(hooks.snapshot()) + assert state["active"] > 0 + assert state["coordinators"] == 1 + native.RuntimeOptions() + assert not any(collection_state(references).values()) + emit("RETAINED") + threading.Event().wait() + else: + if scenario == "callbacks-late": + runtime = None + hooks.release("blocking-provider") + deadline = time.perf_counter_ns() + 2_000_000_000 + if scenario == "callbacks-late": + collected = settle_deferred_references(references, deadline) + assert all(collected.values()), collected + else: + while True: + try: + runtime.shutdown(30_000) + break + except TimeoutError: + assert time.perf_counter_ns() < deadline + time.sleep(0.001) + assert time.perf_counter_ns() < deadline + state = json.loads(hooks.snapshot()) + assert state["active"] == 0 + assert state["cleanupActive"] == 0 + assert state["started"] == state["completed"] + assert state["coordinators"] == 1 + assert state["forceRequests"] == 1 + if runtime is not None: + try: + runtime.start() + raise AssertionError("timed-out runtime restarted") + except RuntimeError as error: + assert "lifecycle_terminal" in str(error) + emit("RECLAIMED") + + +if __name__ == "__main__": + try: + main() + except Exception: + emit("FAILED", error=traceback.format_exc()) + raise diff --git a/tests/lifecycle_probe_support.py b/tests/lifecycle_probe_support.py new file mode 100644 index 0000000..d2d698a --- /dev/null +++ b/tests/lifecycle_probe_support.py @@ -0,0 +1,179 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import json +from pathlib import Path +import queue +import subprocess +import sys +import threading +import time + + +PREFIX = "LIFECYCLE_PROBE " +TAIL_CHARS = 8192 +MAX_RECORD_CHARS = 16_384 +FIRST_STOP_LIMITS = {"late": (6, 7), "retained": (5, 6), "callbacks-late": (5, 6)} +REPEATED_STOP_LIMIT = 1 +TRANSPORT_MARGIN = 2 + + +def timeout_delivery_budget(scenario): + return FIRST_STOP_LIMITS[scenario][1] + REPEATED_STOP_LIMIT + TRANSPORT_MARGIN + + +def assert_timeout_measurements(scenario, message): + lower, upper = FIRST_STOP_LIMITS[scenario] + assert lower <= message["elapsedSeconds"] < upper, message + assert message["repeatedSeconds"] < REPEATED_STOP_LIMIT, message + + +def clock_info(): + clock = time.get_clock_info("perf_counter") + return { + "implementation": clock.implementation, + "resolutionSeconds": clock.resolution, + } + + +class _Tail: + def __init__(self): + self._text = "" + self._lock = threading.Lock() + + def append(self, text): + with self._lock: + self._text = (self._text + text)[-TAIL_CHARS:] + + def text(self): + with self._lock: + return self._text + + +class LifecycleProbe: + def __init__(self, scenario, command=None): + self.scenario = scenario + self.command = command or [ + sys.executable, + str(Path(__file__).parent / "fixtures" / "lifecycle_probe.py"), + scenario, + ] + self.messages = queue.Queue() + self.stdout = _Tail() + self.stderr = _Tail() + self.expected_phase = None + self.last_phase = None + self.last_message = None + self.exit_before_cleanup = None + + def __enter__(self): + self.child = subprocess.Popen( + self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, + ) + self.readers = [ + threading.Thread(target=self._read_stdout, daemon=True), + threading.Thread(target=self._read_stderr, daemon=True), + ] + for reader in self.readers: + reader.start() + return self + + def _read_stdout(self): + continuation = False + try: + while True: + line = self.child.stdout.readline(MAX_RECORD_CHARS + 1) + if not line: + break + self.stdout.append(line) + was_continuation = continuation + continuation = len(line) > MAX_RECORD_CHARS and not line.endswith("\n") + if was_continuation or not line.startswith(PREFIX): + continue + try: + if len(line) > MAX_RECORD_CHARS: + raise ValueError("lifecycle protocol record exceeds the bounded limit") + message = json.loads(line[len(PREFIX):]) + if not isinstance(message, dict) or not isinstance(message.get("phase"), str): + raise ValueError("lifecycle protocol record must contain a string phase") + self.messages.put(("message", message)) + except (ValueError, TypeError) as error: + self.messages.put(("stdout protocol", error)) + except Exception as error: + self.messages.put(("stdout reader", error)) + finally: + self.messages.put(("exit", None)) + + def _read_stderr(self): + try: + while True: + text = self.child.stderr.readline(4096) + if not text: + break + self.stderr.append(text) + except Exception as error: + self.messages.put(("stderr reader", error)) + + def expect_phase(self, phase, timeout): + self.expected_phase = phase + try: + kind, value = self.messages.get(timeout=timeout) + except queue.Empty as error: + raise AssertionError("lifecycle phase delivery watchdog expired") from error + if kind == "exit": + raise AssertionError("child stdout closed before the expected phase") + if kind != "message": + raise AssertionError(f"{kind} failed: {type(value).__name__}: {value}") from value + self.last_message = value + self.last_phase = value["phase"] + assert self.last_phase == phase, f"unexpected lifecycle phase: {self.last_phase}" + return value + + def send(self, message): + self.child.stdin.write(message + "\n") + self.child.stdin.flush() + + def diagnostics(self): + return ( + f"scenario={self.scenario!r} expected_phase={self.expected_phase!r} " + f"last_phase={self.last_phase!r}\n" + f"child_exit_before_cleanup={self.exit_before_cleanup!r} " + f"child_exit={self.child.poll()!r}\n" + f"parent_perf_counter={json.dumps(clock_info())}\n" + f"last_message={json.dumps(self.last_message)[-TAIL_CHARS:]}\n" + f"stdout tail:\n{self.stdout.text()}\n" + f"stderr tail:\n{self.stderr.text()}" + ) + + def __exit__(self, error_type, error, traceback): + self.exit_before_cleanup = self.child.poll() + cleanup_error = None + terminated_for_cleanup = False + try: + try: + self.child.wait(timeout=0.2) + except subprocess.TimeoutExpired: + terminated_for_cleanup = True + self.child.kill() + self.child.wait(timeout=5) + if not terminated_for_cleanup and self.child.returncode != 0: + cleanup_error = AssertionError(f"child exited with status {self.child.returncode}") + except Exception as failure: + cleanup_error = failure + finally: + for reader in self.readers: + reader.join(timeout=2) + if any(reader.is_alive() for reader in self.readers): + cleanup_error = cleanup_error or AssertionError("child stream reader did not stop") + else: + for stream in (self.child.stdin, self.child.stdout, self.child.stderr): + try: + stream.close() + except Exception as failure: + cleanup_error = cleanup_error or failure + if error is not None or cleanup_error is not None: + detail = f"{error}\n" if error is not None else "" + if cleanup_error is not None: + detail += f"child cleanup failed: {cleanup_error}\n" + raise AssertionError(detail + self.diagnostics()) from (error or cleanup_error) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 0000000..23077c0 --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,630 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import gc +import hashlib +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import time +import uuid +import weakref +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from duroxide import Client, Runtime, RuntimeOptions, RuntimeStopObserver, RuntimeStopStatus, SqliteProvider +from duroxide import _duroxide as native +from duroxide import _orchestration_functions +from duroxide.driver import create_generator, next_step, dispose_generator + +from lifecycle_probe_support import ( + LifecycleProbe, + assert_timeout_measurements, + clock_info, + timeout_delivery_budget, +) + + +INSTRUMENTED = os.environ.get("DUROXIDE_LIFECYCLE_TEST_HOOKS") == "1" +SECRET = "Password=lifecycle-secret-sentinel;Host=private-sentinel" +MAX_GRACE_MS = (2**64 - 1) // 1_000_000 - 5000 +hook_test = pytest.mark.skipif(not INSTRUMENTED, reason="requires freshly built test-hooks extension") + + +def fixture(**options): + provider = SqliteProvider.in_memory() + runtime = Runtime(provider, RuntimeOptions(dispatcher_poll_interval_ms=5, **options)) + hooks = native._lifecycle_test_hooks(runtime._native) if INSTRUMENTED else None + return runtime, provider, hooks + + +def diagnostic(error, category): + expected_type = ValueError if category == "lifecycle_invalid_timeout" else RuntimeError + assert type(error.value) is expected_type + message = str(error.value) + assert category in message + assert SECRET not in message + assert "sentinel" not in message + + +def assert_quiescent_shutdown_failure(error): + diagnostic(error, "lifecycle_shutdown_failed") + assert "cleanup completed and is quiescent" in str(error.value) + assert "terminate the process" not in str(error.value) + + +def retired(hooks, forced=None, rollback=False): + state = json.loads(hooks.snapshot()) + assert state["active"] == 0 + assert state["cleanupActive"] == 0 + assert state["started"] == state["completed"] + if rollback: + assert 1 <= state["coordinators"] <= 2 + else: + assert state["coordinators"] == 1 + if forced is not None: + assert state["forceRequests"] == int(forced) + + +def test_native_provenance_and_production_exports(): + extension = Path(native.__file__).resolve() + assert extension.parent == Path(__file__).resolve().parents[1] / "python" / "duroxide" + assert hasattr(native, "_lifecycle_test_hooks") == INSTRUMENTED + assert hasattr(native, "LifecycleTestHooks") == INSTRUMENTED + print("LIFECYCLE_NATIVE " + json.dumps({ + "path": str(extension), + "sha256": hashlib.sha256(extension.read_bytes()).hexdigest(), + "instrumented": INSTRUMENTED, + "pythonVersion": list(sys.version_info[:3]), + "executable": sys.executable, + })) + + +def test_invalid_durations_preserve_runtime_ownership(): + runtime, _, hooks = fixture() + for value in [-1, MAX_GRACE_MS + 1, 2**63 - 1]: + with pytest.raises(ValueError) as error: + runtime.shutdown(value) + diagnostic(error, "lifecycle_invalid_timeout") + for value in [0.5, float("nan"), float("inf"), "0"]: + with pytest.raises(TypeError): + runtime.shutdown(value) + with pytest.raises(OverflowError): + runtime.shutdown(2**63) + runtime.register_orchestration("LifecycleInvalidInput", lambda _ctx, _input: "ok") + assert runtime.start() is None + assert runtime.metrics_snapshot() is not None + with pytest.raises(ValueError): + runtime.shutdown(-1) + assert runtime.metrics_snapshot() is not None + assert runtime.shutdown(0) is None + assert runtime.metrics_snapshot() is None + with pytest.raises(ValueError): + runtime.shutdown(2**63 - 1) + assert runtime.shutdown() is None + if hooks: + retired(hooks) + + +def test_maximum_default_total_boundary_on_no_work_and_repeated_stop(): + runtime, _, _ = fixture() + assert runtime.shutdown(MAX_GRACE_MS) is None + with pytest.raises(ValueError) as error: + runtime.shutdown(MAX_GRACE_MS + 1) + diagnostic(error, "lifecycle_invalid_timeout") + assert runtime.shutdown() is None + + +def test_prestart_shutdown_is_terminal_no_work(): + runtime, _, _ = fixture() + assert runtime.shutdown(0) is None + assert runtime.shutdown() is None + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.register_activity("late", lambda _ctx, _input: "late") + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.register_orchestration("late", lambda _ctx, _input: "late") + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.register_orchestration_versioned("late", "1.0.0", lambda _ctx, _input: "late") + assert runtime.metrics_snapshot() is None + + +def test_missing_native_drivers_reject_before_accepting_startup(): + provider = SqliteProvider.in_memory() + runtime = native.PyRuntime.from_sqlite(provider._native) + with pytest.raises(RuntimeError) as error: + runtime.start() + diagnostic(error, "lifecycle_start_failed") + assert "generator driver not set" in str(error.value) + + runtime.set_generator_driver(create_generator, next_step, dispose_generator) + assert runtime.start() is None + assert runtime.metrics_snapshot() is not None + assert runtime.shutdown(0) is None + assert runtime.shutdown() is None + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + + +def test_no_work_stop_after_missing_native_drivers_is_successful_and_terminal(): + provider = SqliteProvider.in_memory() + runtime = native.PyRuntime.from_sqlite(provider._native) + with pytest.raises(RuntimeError) as error: + runtime.start() + diagnostic(error, "lifecycle_start_failed") + assert runtime.shutdown(0) is None + assert runtime.shutdown() is None + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.set_generator_driver(create_generator, next_step, dispose_generator) + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + + +@pytest.mark.parametrize("decorator", [False, True]) +@pytest.mark.parametrize("versioned", [False, True]) +@pytest.mark.parametrize("terminal_state", ["stopped", "started", "startup-failed"]) +def test_rejected_registrations_preserve_handler_keys_identity_and_execution( + decorator, versioned, terminal_state, +): + active, provider, _ = fixture() + terminal, _, _ = fixture(**({"session_idle_timeout_ms": 0} if terminal_state == "startup-failed" else {})) + client = Client(provider) + prefix = "LifecycleAtomic-" + uuid.uuid4().hex + existing, fresh = prefix + "-existing", prefix + "-fresh" + version = "1.0.0" + existing_key = f"{existing}@{version}" if versioned else existing + owned_keys = {existing, fresh, f"{existing}@{version}", f"{fresh}@{version}"} + saved = {key: _orchestration_functions[key] for key in owned_keys if key in _orchestration_functions} + + def original(_context, value): + return {"handler": "original", "input": value} + + def rejected(_context, value): + return {"handler": "rejected", "input": value} + + try: + if versioned: + active.register_orchestration_versioned(existing, version, original) + else: + active.register_orchestration(existing, original) + active.start() + if terminal_state == "startup-failed": + with pytest.raises(RuntimeError, match="lifecycle_start_failed"): + terminal.start() + elif terminal_state == "started": + terminal.start() + else: + terminal.shutdown(0) + keys_before = {key for key in owned_keys if key in _orchestration_functions} + for index, name in enumerate((existing, fresh)): + register = terminal.register_orchestration_versioned if versioned else terminal.register_orchestration + args = (name, version) if versioned else (name,) + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + if decorator: + register(*args)(rejected) + else: + register(*args, rejected) + assert {key for key in owned_keys if key in _orchestration_functions} == keys_before + assert _orchestration_functions[existing_key] is original + instance = f"{prefix}-{index}" + if versioned: + client.start_orchestration_versioned(instance, existing, index, version) + else: + client.start_orchestration(instance, existing, index) + result = client.wait_for_orchestration(instance, 5000) + assert result.status == "Completed" + assert result.output == {"handler": "original", "input": index} + finally: + try: + active.shutdown(100) + finally: + try: + if terminal_state == "startup-failed": + with pytest.raises(RuntimeError, match="lifecycle_start_failed"): + terminal.shutdown(0) + else: + terminal.shutdown(0) + finally: + for key in owned_keys: + if key in saved: + _orchestration_functions[key] = saved[key] + else: + _orchestration_functions.pop(key, None) + + +def test_fallible_preparation_rejects_options_without_panic(): + runtime, _, _ = fixture(session_idle_timeout_ms=0) + with pytest.raises(RuntimeError) as error: + runtime.start() + diagnostic(error, "lifecycle_start_failed") + with pytest.raises(RuntimeError, match="lifecycle_start_failed"): + runtime.shutdown(0) + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + assert runtime.metrics_snapshot() is None + + +@pytest.mark.parametrize("scenario", ["registry-invalid", "registry-descending"]) +def test_pre_core_startup_failure_is_ordinary_and_retained(scenario): + result = subprocess.run( + [sys.executable, str(Path(__file__).parent / "fixtures" / "lifecycle_probe.py"), scenario], + capture_output=True, text=True, timeout=5, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "REGISTRY_FAILURE_RETAINED" in result.stdout + + +def test_ascending_and_duplicate_versions_keep_existing_successful_startup_policy(): + runtime, _, _ = fixture() + for version in ("1.0.0", "1.0.0", "2.0.0"): + runtime.register_orchestration_versioned("LifecycleVersionPolicy", version, lambda _context, _input: "ok") + assert runtime.start() is None + assert runtime.shutdown(100) is None + assert runtime.shutdown(0) is None + + +def test_idle_shutdown_short_circuits_and_preserves_success(): + runtime, _, hooks = fixture() + runtime.start() + start = time.perf_counter_ns() + assert runtime.shutdown(30_000) is None + elapsed = time.perf_counter_ns() - start + print("LIFECYCLE_IDLE " + json.dumps({"elapsedNanoseconds": elapsed, "clock": clock_info()})) + assert elapsed < 2_000_000_000, clock_info() + assert runtime.shutdown(0) is None + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + if hooks: + retired(hooks, False) + + +def test_sqlite_public_activity_and_orchestration(): + runtime, provider, _ = fixture() + client = Client(provider) + runtime.register_activity("LifecycleEcho", lambda _ctx, value: value) + + @runtime.register_orchestration("LifecycleEchoFlow") + def flow(context, value): + return (yield context.schedule_activity("LifecycleEcho", value)) + + runtime.start() + try: + client.start_orchestration("lifecycle-echo", "LifecycleEchoFlow", "hello") + result = client.wait_for_orchestration("lifecycle-echo", 5000) + assert result.status == "Completed" + assert result.output == "hello" + finally: + runtime.shutdown(100) + + +@pytest.mark.parametrize("typed", [False, True]) +def test_sqlite_races_preserve_failures_values_and_shutdown(typed): + runtime, provider, hooks = fixture() + client = Client(provider) + + @runtime.register_activity("LifecycleRaceFail") + def fail(_context, _input): + raise RuntimeError("race-failure") + + runtime.register_activity("LifecycleRaceValue", lambda _context, _input: {"err": "data", "ok": True}) + + @runtime.register_orchestration("LifecycleRaceFlow") + def flow(context, _input): + schedule = context.schedule_activity_typed if typed else context.schedule_activity + race = context.race_typed if typed else context.race + direct = None + try: + yield schedule("LifecycleRaceFail", None) + except Exception as error: + direct = {"type": type(error).__name__, "message": str(error)} + raced = None + try: + yield race(context.schedule_timer(60_000), schedule("LifecycleRaceFail", None)) + except Exception as error: + raced = {"type": type(error).__name__, "message": str(error)} + winner = yield race(schedule("LifecycleRaceValue", None), context.schedule_timer(60_000)) + return {"direct": direct, "raced": raced, "winner": winner} + + runtime.start() + try: + client.start_orchestration("lifecycle-race", "LifecycleRaceFlow", None) + result = client.wait_for_orchestration("lifecycle-race", 5000) + assert result.status == "Completed" + assert result.output["direct"] is not None + assert "race-failure" in result.output["direct"]["message"] + assert result.output["raced"] == result.output["direct"] + assert result.output["winner"] == {"index": 0, "value": {"err": "data", "ok": True}} + finally: + runtime.shutdown(100) + assert runtime.shutdown(0) is None + if hooks: + retired(hooks) + + +@hook_test +def test_partial_startup_failure_retains_rollback(): + runtime, _, hooks = fixture() + hooks.hold("partial-startup") + hooks.hold("blocking-provider") + hooks.fail("partial-startup", SECRET) + with ThreadPoolExecutor(max_workers=1) as pool: + startup = pool.submit(runtime.start) + try: + hooks.wait_entered("partial-startup") + hooks.wait_entered("blocking-provider") + hooks.release("partial-startup") + with pytest.raises(RuntimeError) as error: + startup.result(timeout=2) + diagnostic(error, "lifecycle_start_failed") + assert json.loads(hooks.snapshot())["active"] > 0 + assert runtime.metrics_snapshot() is None + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + finally: + hooks.release("partial-startup") + hooks.release("blocking-provider") + with pytest.raises(RuntimeError) as error: + runtime.shutdown(0) + assert_quiescent_shutdown_failure(error) + with pytest.raises(RuntimeError, match="lifecycle_shutdown_failed"): + runtime.shutdown(30_000) + retired(hooks, rollback=True) + + +@hook_test +def test_operational_failure_is_retained_on_repeated_shutdown(): + runtime, _, hooks = fixture() + hooks.hold("worker") + hooks.fail("worker", SECRET) + runtime.start() + hooks.wait_entered("worker") + hooks.release("worker") + with pytest.raises(RuntimeError) as error: + runtime.shutdown(100) + assert_quiescent_shutdown_failure(error) + with pytest.raises(RuntimeError, match="lifecycle_shutdown_failed"): + runtime.shutdown(30_000) + retired(hooks, rollback=True) + + +@hook_test +def test_cancellation_safe_fetch_retires_without_waiting_for_its_gate(): + runtime, _, hooks = fixture() + hooks.hold("provider") + runtime.start() + hooks.wait_entered("provider") + try: + runtime.shutdown(1000) + retired(hooks, False) + finally: + hooks.release("provider") + + +@hook_test +@pytest.mark.parametrize("force", [False, True]) +def test_ordered_force_and_retirement_100_times(force): + with ThreadPoolExecutor(max_workers=1) as pool: + for _ in range(100): + runtime, _, hooks = fixture() + hooks.hold("blocking-provider") + hooks.hold("force" if force else "coordinator") + runtime.start() + hooks.wait_entered("blocking-provider") + stop = pool.submit(runtime.shutdown, 0 if force else 30_000) + try: + hooks.wait_entered("force" if force else "coordinator") + assert json.loads(hooks.snapshot())["active"] > 0 + finally: + hooks.release("blocking-provider") + hooks.release("force" if force else "coordinator") + assert stop.result(timeout=2) is None + assert runtime.shutdown() is None + retired(hooks, force) + + +@hook_test +@pytest.mark.parametrize("scenario", ["late", "retained", "callbacks-late"]) +def test_isolated_non_yielding_provider_timeout(scenario): + with LifecycleProbe(scenario) as probe: + probe.expect_phase("STOPPING", 15) + timed_out = probe.expect_phase("TIMED_OUT", timeout_delivery_budget(scenario)) + print("LIFECYCLE_TIMING " + json.dumps({"scenario": scenario, **timed_out})) + assert_timeout_measurements(scenario, timed_out) + probe.send("retain" if scenario == "retained" else "release") + probe.expect_phase("RETAINED" if scenario == "retained" else "RECLAIMED", 3) + + +@pytest.mark.parametrize("scenario", [ + "callbacks-prestart", "callbacks-stopped", "callbacks-preparation-failed", +]) +def test_activity_only_callbacks_collect_after_quiescence_in_live_process(scenario): + database_directory = Path(__file__).resolve().parents[1] / "target" / "lifecycle-callbacks" / uuid.uuid4().hex + database_directory.mkdir(parents=True) + command = [ + sys.executable, str(Path(__file__).parent / "fixtures" / "lifecycle_probe.py"), + scenario, str(database_directory), + ] + try: + with LifecycleProbe(scenario, command) as probe: + report = probe.expect_phase("COLLECTION", 10) + print("LIFECYCLE_COLLECTION " + json.dumps({"scenario": scenario, **report})) + assert probe.child.poll() is None + assert set(report["cases"]) == {"noncapturing", "bound", "self-capturing"} + for case in report["cases"].values(): + assert all(case["collected"].values()), report + probe.send("finish") + probe.expect_phase("COLLECTION_FINISHED", 3) + finally: + shutil.rmtree(database_directory) + + +@hook_test +def test_activity_only_callbacks_collect_after_quiescent_operational_failure(): + test_activity_only_callbacks_collect_after_quiescence_in_live_process("callbacks-operational-failed") + + +def test_retained_completed_runtime_allows_natural_process_exit(): + result = subprocess.run( + [sys.executable, str(Path(__file__).parent / "fixtures" / "lifecycle_probe.py"), "exit"], + capture_output=True, text=True, timeout=3, check=True, + ) + assert "EXIT_READY" in result.stdout + + +def test_stop_observer_precore_validation_and_timeout_do_not_activate_or_stop(): + runtime, _, hooks = fixture() + observer = runtime.stop_observer() + not_requested = RuntimeStopStatus.NOT_REQUESTED + assert observer.state is not_requested + assert observer.wait_for_change(RuntimeStopStatus.FAULTED, 0) is not_requested + with pytest.raises(TypeError, match="Runtime.stop_observer"): + RuntimeStopObserver() + with pytest.raises(AttributeError): + observer.state = RuntimeStopStatus.REQUESTED + for value in [-1, (2**64 - 1) // 1_000_000 + 1, 2**63 - 1]: + with pytest.raises(ValueError, match="stop_observation_invalid_timeout"): + observer.wait_for_change(not_requested, value) + for value in [0.5, float("nan"), float("inf"), "0"]: + with pytest.raises(TypeError): + observer.wait_for_change(not_requested, value) + with pytest.raises(OverflowError): + observer.wait_for_change(not_requested, 2**63) + for value in ["Unknown", "", "requested"]: + with pytest.raises(ValueError, match="stop_observation_invalid_status"): + observer.wait_for_change(value, 0) + for value in [None, 1, False]: + with pytest.raises(TypeError): + observer.wait_for_change(value, 0) + for _ in range(2): + with pytest.raises(TimeoutError, match="stop_observation_timed_out"): + observer.wait_for_change(not_requested, 0) + assert observer.state is not_requested + if hooks: + assert json.loads(hooks.snapshot())["started"] == 0 + observer.close() + observer.close() + with pytest.raises(RuntimeError, match="stop_observation_closed"): + _ = observer.state + with pytest.raises(RuntimeError, match="stop_observation_closed"): + observer.wait_for_change(not_requested) + runtime.register_activity("ObserverDidNotStop", lambda _context, value: value) + runtime.start() + runtime.shutdown(100) + + +def test_stop_observer_precore_stop_preserves_final_state_for_independent_late_readers(): + runtime, _, _ = fixture() + observer, closed = runtime.stop_observer(), runtime.stop_observer() + with pytest.raises(ValueError): + runtime.shutdown(-1) + assert observer.state is RuntimeStopStatus.NOT_REQUESTED + closed.close() + runtime.shutdown(0) + late = runtime.stop_observer() + for reader in (observer, late): + assert reader.state is RuntimeStopStatus.REQUESTED + assert reader.wait_for_change(RuntimeStopStatus.NOT_REQUESTED, 0) is RuntimeStopStatus.REQUESTED + assert reader.wait_for_change(RuntimeStopStatus.REQUESTED, 0) is None + assert reader.state is RuntimeStopStatus.REQUESTED + reader.close() + with pytest.raises(RuntimeError, match="lifecycle_terminal"): + runtime.start() + + +@pytest.mark.parametrize("failure", ["options", "registry"]) +def test_stop_observer_precore_startup_error_is_not_a_runtime_fault(failure): + runtime, _, _ = fixture(**({"session_idle_timeout_ms": 0} if failure == "options" else {})) + if failure == "registry": + runtime.register_orchestration_versioned("ObserverInvalidRegistry", "invalid", lambda _context, _input: "ok") + observer = runtime.stop_observer() + with pytest.raises(RuntimeError, match="lifecycle_start_failed") as start: + runtime.start() + assert observer.state is RuntimeStopStatus.NOT_REQUESTED + with pytest.raises(RuntimeError) as stop: + runtime.shutdown(0) + assert str(start.value) == str(stop.value) + assert observer.wait_for_change(RuntimeStopStatus.NOT_REQUESTED, 0) is RuntimeStopStatus.REQUESTED + assert observer.wait_for_change(RuntimeStopStatus.REQUESTED, 0) is None + observer.close() + + +def test_stop_observer_core_cleanup_is_not_source_end(): + runtime, _, _ = fixture() + observer = runtime.stop_observer() + runtime.start() + assert observer.state is RuntimeStopStatus.NOT_REQUESTED + runtime.shutdown(100) + assert observer.state is RuntimeStopStatus.REQUESTED + with pytest.raises(TimeoutError): + observer.wait_for_change(RuntimeStopStatus.REQUESTED, 0) + reference = weakref.ref(runtime) + del runtime + gc.collect() + assert reference() is None + assert observer.wait_for_change(RuntimeStopStatus.REQUESTED, 1000) is None + assert observer.state is RuntimeStopStatus.REQUESTED + observer.close() + + +def run_observer_probe(scenario): + directory = Path(__file__).resolve().parents[1] / "target" / "lifecycle-observers" / uuid.uuid4().hex + directory.mkdir(parents=True) + command = [ + sys.executable, str(Path(__file__).parent / "fixtures" / "lifecycle_probe.py"), + "observer-" + scenario, str(directory), + ] + try: + with LifecycleProbe("observer-" + scenario, command) as probe: + report = probe.expect_phase("OBSERVER_RESULT", 45 if "races" in scenario or "faults" in scenario else 12) + print("LIFECYCLE_OBSERVER " + json.dumps(report)) + assert probe.child.poll() is None + probe.send("finish") + probe.expect_phase("OBSERVER_FINISHED", 3) + assert probe.child.wait(timeout=3) == 0 + return report + finally: + shutil.rmtree(directory) + + +@pytest.mark.parametrize("scenario", [ + "timeout", "close", "interrupt", "signal-close", "signal-error", "signal-return", +]) +def test_stop_observer_interrupts_before_held_callback_cleanup_in_both_modes(scenario): + report = run_observer_probe(scenario) + assert report["otherThreadProgress"] and report["nativeWaitActive"] + assert report["callbackHeldAtSettlement"] + assert report["stopStateAtSettlement"] == RuntimeStopStatus.NOT_REQUESTED + + +@pytest.mark.parametrize("scenario", ["ordered-races", "source-end", "collection", "exit"]) +def test_stop_observer_owned_lifetime_and_ordered_races_in_both_modes(scenario): + report = run_observer_probe(scenario) + if scenario == "ordered-races": + assert report["iterations"] == 100 + assert report["duplicateSettlements"] == 0 + assert report["outcomes"] == {"change": 25, "close": 25, "timeout": 25, "precore": 25} + if "collected" in report: + assert all(report["collected"].values()), report + + +@hook_test +@pytest.mark.parametrize("scenario", [ + "attachment-faults", "fault-before-stop", "source-freeze", "delivery-retention", "transport-failure", +]) +def test_stop_observer_real_faults_and_native_notification_lifetime(scenario): + report = run_observer_probe(scenario) + if scenario == "attachment-faults": + assert report["iterations"] == 100 + assert report["faultObservedBeforeCleanup"] + elif scenario == "fault-before-stop": + assert report["faultWithoutSdkStop"] and report["callbackHeldAtFault"] + assert report["lateState"] == RuntimeStopStatus.FAULTED + assert report["callbackRetiredAfterRelease"] + elif scenario == "delivery-retention": + assert report["nativeTailHeld"] and report["oldDeliveryDidNotSettleRetry"] + assert all(report["collected"].values()), report diff --git a/tests/test_lifecycle_probe_support.py b/tests/test_lifecycle_probe_support.py new file mode 100644 index 0000000..4320518 --- /dev/null +++ b/tests/test_lifecycle_probe_support.py @@ -0,0 +1,144 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import json +import queue +import sys + +import pytest + +from lifecycle_probe_support import ( + LifecycleProbe, + TAIL_CHARS, + assert_timeout_measurements, + timeout_delivery_budget, +) + + +def child_command(program): + return [sys.executable, "-u", "-c", program] + + +def assert_child_cleaned_up(probe): + assert probe.child.poll() is not None + assert all(not reader.is_alive() for reader in probe.readers) + assert all(stream.closed for stream in (probe.child.stdin, probe.child.stdout, probe.child.stderr)) + + +def test_deliberate_child_failure_preserves_bounded_evidence(): + program = """ +import json, sys, traceback +print("old-stdout-" + "x" * 40000, flush=True) +print("stdout-tail-marker", flush=True) +print("old-stderr-" + "y" * 40000, file=sys.stderr, flush=True) +print("stderr-tail-marker", file=sys.stderr, flush=True) +print('LIFECYCLE_PROBE {"phase":"STOPPING"}', flush=True) +try: + raise RuntimeError("deliberate-child-failure") +except RuntimeError: + traceback.print_exc() + print("LIFECYCLE_PROBE " + json.dumps({ + "phase": "FAILED", "error": traceback.format_exc(), + }), flush=True) +sys.exit(17) +""" + probe = LifecycleProbe("deliberate-failure", child_command(program)) + with pytest.raises(AssertionError) as failure: + with probe: + probe.expect_phase("STOPPING", 5) + probe.expect_phase("TIMED_OUT", 5) + evidence = str(failure.value) + for expected in ( + "scenario='deliberate-failure'", "expected_phase='TIMED_OUT'", + "last_phase='FAILED'", "child_exit=17", "stdout-tail-marker", + "stderr-tail-marker", "Traceback", "deliberate-child-failure", + "parent_perf_counter=", "implementation", "resolutionSeconds", + ): + assert expected in evidence + assert "old-stdout-" not in evidence + assert "old-stderr-" not in evidence + assert len(probe.stdout.text()) <= TAIL_CHARS + assert len(probe.stderr.text()) <= TAIL_CHARS + assert_child_cleaned_up(probe) + + +def test_malformed_child_protocol_preserves_original_reader_error(): + probe = LifecycleProbe("malformed", child_command( + 'print(\'LIFECYCLE_PROBE {"phase":"STOPPING"}\', flush=True)\n' + 'print("LIFECYCLE_PROBE {broken-json", flush=True)\n' + )) + with pytest.raises(AssertionError) as failure: + with probe: + probe.expect_phase("STOPPING", 5) + probe.expect_phase("TIMED_OUT", 5) + assert "stdout protocol failed: JSONDecodeError" in str(failure.value) + assert "last_phase='STOPPING'" in str(failure.value) + assert isinstance(failure.value.__cause__.__cause__, json.JSONDecodeError) + assert_child_cleaned_up(probe) + + +def test_delivery_watchdog_terminates_child_and_preserves_timeout_cause(): + probe = LifecycleProbe("silent-child", child_command( + 'import threading\n' + 'print(\'LIFECYCLE_PROBE {"phase":"STOPPING"}\', flush=True)\n' + 'threading.Event().wait()\n' + )) + with pytest.raises(AssertionError) as failure: + with probe: + probe.expect_phase("STOPPING", 5) + probe.expect_phase("TIMED_OUT", 0.05) + assert "phase delivery watchdog expired" in str(failure.value) + assert "last_phase='STOPPING'" in str(failure.value) + assert isinstance(failure.value.__cause__.__cause__, queue.Empty) + assert_child_cleaned_up(probe) + + +def test_child_failure_after_expected_phase_is_not_hidden_by_cleanup(): + probe = LifecycleProbe("failed-exit", child_command( + 'import sys\n' + 'print(\'LIFECYCLE_PROBE {"phase":"RECLAIMED"}\', flush=True)\n' + 'sys.exit(19)\n' + )) + with pytest.raises(AssertionError, match="child exited with status 19") as failure: + with probe: + probe.expect_phase("RECLAIMED", 5) + assert "child_exit=19" in str(failure.value) + assert_child_cleaned_up(probe) + + +def test_oversized_protocol_is_reported_while_both_streams_are_drained(): + probe = LifecycleProbe("oversized", child_command( + 'import sys\n' + 'print("LIFECYCLE_PROBE " + "x" * 40000, flush=True)\n' + 'print("stderr-drained", file=sys.stderr, flush=True)\n' + )) + with pytest.raises(AssertionError, match="record exceeds the bounded limit") as failure: + with probe: + probe.expect_phase("STOPPING", 5) + assert "stderr-drained" in str(failure.value) + assert isinstance(failure.value.__cause__.__cause__, ValueError) + assert_child_cleaned_up(probe) + + +@pytest.mark.parametrize("scenario,elapsed", [("late", 6.8), ("retained", 5.8)]) +def test_delivery_watchdog_includes_both_observations_and_transport(scenario, elapsed): + message = {"elapsedSeconds": elapsed, "repeatedSeconds": 0.5} + assert_timeout_measurements(scenario, message) + first_upper = 7 if scenario == "late" else 6 + assert timeout_delivery_budget(scenario) == first_upper + 1 + 2 + assert timeout_delivery_budget(scenario) > elapsed + message["repeatedSeconds"] + + +@pytest.mark.parametrize("scenario,elapsed,repeated", [ + ("late", 5.999999999, 0.1), + ("late", 7, 0.1), + ("retained", 4.999999999, 0.1), + ("retained", 6, 0.1), + ("late", 6, 1), + ("retained", 5, 1), +]) +def test_transport_headroom_does_not_relax_measurement_limits(scenario, elapsed, repeated): + with pytest.raises(AssertionError): + assert_timeout_measurements( + scenario, {"elapsedSeconds": elapsed, "repeatedSeconds": repeated}, + )