From 81711b88a224bb54c2448c48f7951350d4c334c7 Mon Sep 17 00:00:00 2001 From: thc1006 <84045975+thc1006@users.noreply.github.com> Date: Wed, 9 Sep 2026 23:53:45 +0800 Subject: [PATCH] BUG: make a failing Monte Carlo worker say so Six ways a worker that failed got away without saying so. It died inside its own handler on an unbound name, so the manager lock it was holding was never given back and the run waited for good. A worker that was killed ran no handler, set no event, and left only an exit code nobody read, so simulate() returned normally with rows missing. Reporting a failure could itself block on a lock a dead sibling still held. Both names are bound before the try now, the handler reports through a bounded lock and releases it from a finally, and the parent reads exit codes and the failure event instead of joining unbounded. A run that is only slow is still never cut short: how a worker ended decides that, not how long it took. An exit code cannot show a worker that left between claiming an index and recording it, so a run is checked against its own logs at the end. Both must hold exactly the simulations asked for, none twice, none unreadable. Scope is the parallel producer. __run_in_serial has the same unbound name and belongs to #1177. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com> --- CHANGELOG.md | 1 + rocketpy/simulation/monte_carlo.py | 317 ++++++++++++-- .../test_monte_carlo_parallel_runs.py | 1 + .../test_monte_carlo_run_completeness.py | 272 ++++++++++++ .../test_monte_carlo_worker_exit.py | 78 ++++ .../test_monte_carlo_worker_join.py | 184 ++++++++ .../test_monte_carlo_worker_reporting.py | 406 ++++++++++++++++++ 7 files changed, 1229 insertions(+), 30 deletions(-) create mode 100644 tests/unit/simulation/test_monte_carlo_run_completeness.py create mode 100644 tests/unit/simulation/test_monte_carlo_worker_exit.py create mode 100644 tests/unit/simulation/test_monte_carlo_worker_join.py create mode 100644 tests/unit/simulation/test_monte_carlo_worker_reporting.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 31e66e333..933226147 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -62,6 +62,7 @@ Attention: The newest changes should be on top --> ### Fixed - BUG: Correct the gravity sign an `Accelerometer` applies when `consider_gravity=True`. The gravitational field was added to the inertial acceleration instead of subtracted from it, so the sensor reported the negative of the proper acceleration along the vertical: one at rest read -g rather than +g. Recorded accelerometer data taken with `consider_gravity=True` changes sign in that term. [#1175](https://github.com/RocketPy-Team/RocketPy/pull/1175) +- BUG: Report a Monte Carlo worker that fails instead of hanging or passing for a finished run [#1182](https://github.com/RocketPy-Team/RocketPy/pull/1182) - BUG: Sample `StochasticFlight` inputs once per simulation [#1126](https://github.com/RocketPy-Team/RocketPy/pull/1126) [#1090](https://github.com/RocketPy-Team/RocketPy/issues/1090) - BUG: Fix spurious `ValueError` from floating-point roundoff at exact tank depletion [#1166](https://github.com/RocketPy-Team/RocketPy/pull/1166) - BUG: Draw each declared eccentricity once per simulation [#1168](https://github.com/RocketPy-Team/RocketPy/pull/1168) diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py index c2dcd4030..8c00c1385 100644 --- a/rocketpy/simulation/monte_carlo.py +++ b/rocketpy/simulation/monte_carlo.py @@ -18,9 +18,10 @@ import os import traceback import warnings +from contextlib import suppress from numbers import Real from pathlib import Path -from time import time +from time import monotonic, time import numpy as np import simplekml @@ -43,6 +44,19 @@ # this is the only format it can both resume from and overwrite safely. _SIMULATION_LOG_SUFFIX = ".txt" +# Which simulation a row belongs to. Every check on a finished run reads it. +_SIMULATION_INDEX_KEY = "index" + +# How a manager that has gone away answers a proxy call. +_MANAGER_IS_GONE = (OSError, EOFError) + +# Bounded, so a lock its dead holder never released cannot pin this worker. +_REPORT_LOCK_SECONDS = 5.0 + +# Longer than the exit-code path: a worker that only read the event is healthy +# and leaving at the end of the simulation in hand, not blocked on a dead lock. +_REPORTED_FAILURE_GRACE_SECONDS = 60.0 + def _refuse_logs_this_run_cannot_write( input_file, output_file, error_file, export_config=None @@ -300,6 +314,14 @@ def simulate( ------- None + Raises + ------ + RuntimeError + If a parallel run does not finish. A worker that ends badly, one + that reports a failure, and logs that do not hold every simulation + asked for are each refused, since a run that lost work must not be + reported as one that completed. + Notes ----- If you need to stop the simulations after starting them, you can @@ -471,22 +493,26 @@ def __run_in_parallel(self, n_workers=None): processes = [] seeds = np.random.SeedSequence().spawn(n_workers) - for seed in seeds: - sim_producer = multiprocess.Process( - target=self.__sim_producer, - args=( - seed, - sim_monitor, - mutex, - simulation_error_event, - ), - ) - processes.append(sim_producer) - sim_producer.start() - try: - for sim_producer in processes: - sim_producer.join() + for seed in seeds: + sim_producer = multiprocess.Process( + target=self.__sim_producer, + args=( + seed, + sim_monitor, + mutex, + simulation_error_event, + ), + ) + sim_producer.start() + # Started first: one that never did cannot be joined, and + # a later start failing still has to bring these down. + processes.append(sim_producer) + + _join_the_workers(processes, simulation_error_event) + + # Before the event: a killed worker never sets it. + _refuse_a_worker_that_did_not_finish(processes) # Handle error from the child processes if simulation_error_event.is_set(): @@ -496,15 +522,21 @@ def __run_in_parallel(self, n_workers=None): "for more information." ) + # An exit code cannot show a worker that left between + # claiming an index and recording it. + _refuse_logs_missing_a_simulation( + self.input_file, self.output_file, self.number_of_simulations + ) + sim_monitor.print_final_status() # Handle error from the main process # pylint: disable=broad-except except (Exception, KeyboardInterrupt) as error: - simulation_error_event.set() - - for sim_producer in processes: - sim_producer.join() + # Bounded here too. An unbounded join undid the bound above. + _stop_the_workers_still_running( + processes, simulation_error_event, _SHUTDOWN_GRACE_SECONDS + ) if not isinstance(error, KeyboardInterrupt): raise error @@ -531,6 +563,8 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disa error_event : multiprocess.Event Event signaling an error occurred during the simulation. """ + # The handler reads both, and a failure above the loop precedes them. + sim_idx, inputs_json = None, "" try: # Ensure Processes generate different random numbers self.environment._set_stochastic(seed) @@ -567,18 +601,48 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disa finally: mutex.release() - except Exception: # pylint: disable=broad-except - mutex.acquire() - with open(self.error_file, "a", encoding="utf-8") as f: - f.write(inputs_json) + # Nothing is in flight between two simulations, nor are these. + sim_idx, inputs_json = None, "" - # See note above: must use print() to remain visible from a - # multiprocessing worker process. - _SimMonitor.reprint( - f"Error on iteration {sim_idx}:\n{traceback.format_exc()}" - ) + except Exception: # pylint: disable=broad-except + if not self.__report_a_failed_simulation( + sim_idx, inputs_json, mutex, error_event + ): + # The event could not be set; the exit code is what is left. + raise + + def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event): + """Write down and announce a simulation this worker could not finish. + + The event goes first and from outside the lock, since a worker that + cannot write its diagnostics still has to be able to stop the others. + Each step under the lock is suppressed on its own: a full disk would + otherwise replace the failure being reported, and the lock is a + manager's, so ending while holding it leaves the next worker waiting + on a process that no longer exists. + """ + details = traceback.format_exc() + where = "worker startup" if sim_idx is None else f"iteration {sim_idx}" + announced = False + with suppress(_MANAGER_IS_GONE): error_event.set() - mutex.release() + announced = True + + held = False + with suppress(*_MANAGER_IS_GONE): + held = mutex.acquire(timeout=_REPORT_LOCK_SECONDS) + try: + with suppress(OSError): + with open(self.error_file, "a", encoding="utf-8") as f: + f.write(_worker_failure_record(where, details, inputs_json)) + with suppress(OSError, ValueError): + # Must use print() to remain visible from a worker process. + _SimMonitor.reprint(f"Error on {where}:\n{details}") + finally: + if held: + with suppress(*_MANAGER_IS_GONE): + mutex.release() + return announced def __run_single_simulation(self): """Runs a single simulation and returns the inputs and outputs. @@ -983,6 +1047,13 @@ def _check_data_collector(self, data_collector): "Invalid 'data_collector' key! " f"Variable names overwrites 'export_list' key '{key}'." ) + if key == _SIMULATION_INDEX_KEY: + raise ValueError( + f"Invalid 'data_collector' key '{key}'! It is the " + f"number of the simulation the row belongs to, which " + f"is written after the collectors run and cannot be " + f"replaced by one." + ) if not callable(callback): raise ValueError( f"Invalid value in 'data_collector' for key '{key}'! " @@ -1755,6 +1826,192 @@ def export_errors_to_json(self, filename): self._write_log_to_json(self.errors_log, filename) +# Prompt enough to notice a dead worker, cheap enough over a run of hours. +_JOIN_POLL_SECONDS = 0.2 +_SHUTDOWN_GRACE_SECONDS = 5.0 + + +def _ended_badly(worker): + """Whether a worker has stopped, and stopped for the wrong reason.""" + return worker.exitcode not in (None, 0) + + +def _a_failure_was_reported(error_event): + """Whether a worker has said it failed, false if it cannot be asked.""" + with suppress(*_MANAGER_IS_GONE): + return error_event.is_set() + return False + + +def _wait_for_the_workers(processes, seconds): + """Join every worker against one shared deadline, not one each. + + Monotonic, since a clock correction would move a wall-clock deadline. + """ + deadline = monotonic() + seconds + for worker in processes: + worker.join(timeout=max(0.0, deadline - monotonic())) + + +def _stop_the_workers_still_running(processes, error_event, grace_period): + """Ask the rest to stop, end what cannot, kill what outlives that. + + Asked first because a worker between simulations reads the event and leaves + with its logs intact. One blocked on a lock its dead sibling was holding + never reaches that check. Terminate runs no handlers, so it comes second, + and a worker can still ignore it. + """ + with suppress(_MANAGER_IS_GONE): + error_event.set() + _wait_for_the_workers(processes, grace_period) + + for worker in processes: + if worker.is_alive(): + worker.terminate() + _wait_for_the_workers(processes, grace_period) + + for worker in processes: + if worker.is_alive(): + worker.kill() + _wait_for_the_workers(processes, grace_period) + + +def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS): + """Wait for the workers, and stop once one of them has failed. + + A reported failure ends the wait as well as a bad exit code, since a + worker that reports one leaves cleanly and says nothing through its exit + status. Its siblings read the event between simulations, but one blocked + on a lock nobody owns never reaches that check, and the run is already + short a simulation either way, so the wait is bounded here rather than + left to them. The reported path gets the longer grace: those siblings are + working, not stuck. + + Slowness alone ends nothing. With no failure reported a healthy worker is + given as long as it needs. + """ + while any(worker.is_alive() for worker in processes): + for worker in processes: + worker.join(timeout=_JOIN_POLL_SECONDS) + if any(_ended_badly(worker) for worker in processes): + _stop_the_workers_still_running(processes, error_event, grace_period) + return + if _a_failure_was_reported(error_event): + _stop_the_workers_still_running( + processes, error_event, _REPORTED_FAILURE_GRACE_SECONDS + ) + return + + +def _worker_failure_record(where, details, inputs_json=""): + """A row saying what failed, and what the simulation had drawn so far. + + The inputs alone left the error file with no stage and no traceback, which + is what the caller is sent there to read. + """ + record = {"index": None, "stage": where, "error": details} + with suppress(ValueError): + drawn = json.loads(inputs_json) + if isinstance(drawn, dict): + record["index"] = drawn.get("index") + record["inputs"] = drawn + return json.dumps(record) + "\n" + + +def _indices_a_log_holds(path): + """Every index a log records, in order, and ``None`` for a row it cannot.""" + found = [] + with open(path, "r", encoding="utf-8") as recorded: + for line in recorded: + if not line.strip(): + continue + try: + index = json.loads(line)["index"] + except (ValueError, KeyError, TypeError): + found.append(None) + continue + usable = ( + isinstance(index, int) and not isinstance(index, bool) and index >= 0 + ) + found.append(index if usable else None) + return found + + +def _refuse_logs_missing_a_simulation(input_file, output_file, target): + """Raise unless both logs hold every simulation the run was asked for. + + An exit code says how a worker ended, never whether the index it had + already claimed reached the logs, and the monitor counts claims rather than + rows. A worker that leaves between the two is invisible to everything else + here, so the logs themselves are what the run is judged on. + + Rows numbered past the target are left alone: an append given a smaller + target than the checkpoint already holds is an append question, not a lost + simulation. What each log holds still has to be the consecutive run it + claims to be, so its indices are required to be exactly as many as its + rows, which refuses a stray number and a hole without needing to be told + how long the checkpoint was. The two logs must also agree row for row, + since a record goes into both under one lock. Streamed rather than read + through ``_read_log_file``, which would hold every row in memory. + """ + wanted = set(range(target)) + recorded = {} + for label, path in (("input", input_file), ("output", output_file)): + found = _indices_a_log_holds(path) + recorded[label] = found + held = set(found) + if None in held: + raise RuntimeError( + f"The run is incomplete: the {label} log has rows that cannot " + f"be read, so what it holds cannot be established." + ) + if len(found) != len(held): + raise RuntimeError( + f"The run is incomplete: the {label} log records " + f"{len(found) - len(held)} simulation(s) more than once." + ) + missing = sorted(wanted - held) + if missing: + raise RuntimeError( + f"The run is incomplete: the {label} log is missing " + f"{len(missing)} of {target} simulations, the first being " + f"{missing[0]}." + ) + strays = sorted(held - set(range(len(found)))) + if strays: + raise RuntimeError( + f"The run is incomplete: the {label} log numbers a simulation " + f"{strays[0]}, past the {len(found)} it holds, so what it " + f"records is not one run of consecutive simulations." + ) + + if recorded["input"] != recorded["output"]: + raise RuntimeError( + "The run is incomplete: the input and output logs do not record " + "the same simulations in the same order. A record is written to " + "both under one lock, so they hold two different runs." + ) + + +def _refuse_a_worker_that_did_not_finish(processes): + """Raise if any worker left without exiting cleanly. + + A negative code is the signal that ended it, ``None`` one still running. + """ + unfinished = [ + f"worker {position} with exit code {process.exitcode}" + for position, process in enumerate(processes) + if process.exitcode != 0 + ] + if not unfinished: + return + raise RuntimeError( + f"The run is incomplete: {', '.join(unfinished)}. A worker that ends " + "this way records nothing and cannot say why, so the simulations it " + "held are missing from the results." + ) + + def _import_multiprocess(): """Import the necessary modules and submodules for the multiprocess library. diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py index 4ab0be440..21ac90589 100644 --- a/tests/unit/simulation/test_monte_carlo_parallel_runs.py +++ b/tests/unit/simulation/test_monte_carlo_parallel_runs.py @@ -7,6 +7,7 @@ def test_a_monte_carlo_run_finishes( stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel ): + """A real run completes and records every simulation, both modes.""" # The parallel path hands each worker a SeedSequence rather than an int, and # nothing else in the suite exercises that. A worker that dies on it is not # reported, so this reads as a hang rather than as a failure. diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py new file mode 100644 index 000000000..d98f51390 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py @@ -0,0 +1,272 @@ +import ast +import inspect +import json +import os + +import pytest + +from rocketpy.simulation import monte_carlo as mc_module +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _refuse_logs_missing_a_simulation, +) + + +def _a_log(tmp_path, name, rows): + path = tmp_path / name + path.write_text("".join(rows), encoding="utf-8") + return str(path) + + +def _row(index): + return json.dumps({"index": index, "mass": 1.0}) + "\n" + + +def _complete(tmp_path, count=3, name="ok"): + rows = [_row(index) for index in range(count)] + return ( + _a_log(tmp_path, f"{name}.inputs.txt", rows), + _a_log(tmp_path, f"{name}.outputs.txt", rows), + ) + + +def test_a_run_that_recorded_everything_is_accepted(tmp_path): + """Logs holding every index the run asked for raise nothing.""" + inputs, outputs = _complete(tmp_path) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_blank_lines_between_rows_are_not_simulations(tmp_path): + """A blank line is skipped rather than counted as an unreadable row.""" + # An interrupted write leaves them, and reading one as a row would report + # a damaged log for a run that lost nothing. + rows = [_row(0), "\n", _row(1), " \n", _row(2)] + inputs = _a_log(tmp_path, "gappy.inputs.txt", rows) + outputs = _a_log(tmp_path, "gappy.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_missing_simulation_is_refused(tmp_path): + """A gap in the output log names the first index that is missing.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.outputs.txt", [_row(0), _row(2)]) + + with pytest.raises(RuntimeError, match=r"output log.*missing.*being 1"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_simulation_recorded_twice_is_refused(tmp_path): + """A duplicated index is refused, since a set alone would hide it.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(1), _row(2)]) + + with pytest.raises(RuntimeError, match="more than once"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_row_that_cannot_be_read_is_refused(tmp_path): + """A torn row means the log's contents cannot be established.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.outputs.txt", [_row(0), "{half a row\n", _row(2)]) + + with pytest.raises(RuntimeError, match="cannot be read"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_rows_numbered_past_the_run_are_left_alone(tmp_path): + """An append below what a checkpoint already holds loses no simulation.""" + # Refusing these said rows were missing when they were extra, and moved + # what append means inside a change about worker failure. + rows = [_row(index) for index in range(4)] + inputs = _a_log(tmp_path, "big.inputs.txt", rows) + outputs = _a_log(tmp_path, "big.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_logs_that_hold_different_simulations_are_refused(tmp_path): + """The input and output logs have to hold the same indices.""" + inputs = _a_log(tmp_path, "a.inputs.txt", [_row(0), _row(1)]) + outputs = _a_log(tmp_path, "a.outputs.txt", [_row(0), _row(2)]) + + with pytest.raises(RuntimeError): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +@pytest.mark.parametrize("index", [True, False, 1.0, -1, [], "1", None]) +def test_an_index_that_is_not_a_whole_number_is_refused(tmp_path, index): + """Only a non-negative int names a simulation, whatever compares equal.""" + # True and 1.0 both equal 1, so either could stand in for a simulation that + # was never run. An unhashable one used to escape as a raw TypeError. + rows = [_row(0), _row(index)] + inputs = _a_log(tmp_path, "odd.inputs.txt", rows) + outputs = _a_log(tmp_path, "odd.outputs.txt", rows) + + with pytest.raises(RuntimeError, match="cannot be read"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_a_stray_index_past_the_rows_is_refused(tmp_path): + """A number no run of this length could have produced is not a simulation.""" + # A worker that claimed one index too many writes exactly this, and the + # target alone cannot tell it from a checkpoint that is legitimately longer. + rows = [_row(0), _row(1), _row(99)] + inputs = _a_log(tmp_path, "stray.inputs.txt", rows) + outputs = _a_log(tmp_path, "stray.outputs.txt", rows) + + with pytest.raises(RuntimeError, match="past the 3 it holds"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_the_order_a_parallel_run_finishes_in_is_not_a_hole(tmp_path): + """Rows arrive in completion order, which is not sorted and not wrong.""" + rows = [_row(1), _row(0), _row(2)] + inputs = _a_log(tmp_path, "para.inputs.txt", rows) + outputs = _a_log(tmp_path, "para.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_logs_that_disagree_on_the_order_are_refused(tmp_path): + """A record goes into both logs under one lock, so the order is the same.""" + inputs = _a_log(tmp_path, "order.inputs.txt", [_row(0), _row(1)]) + outputs = _a_log(tmp_path, "order.outputs.txt", [_row(1), _row(0)]) + + with pytest.raises(RuntimeError, match="same order"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def _leave_cleanly_without_recording(_flight): + # A worker that ends the way an out-of-memory kill ends it, but with the + # status of one that finished. Nothing about the process says otherwise. + os._exit(0) + + +def test_a_worker_that_leaves_cleanly_without_recording_is_not_a_success( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """A zero exit with no row written makes ``simulate`` raise.""" + analysis = MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"leave": _leave_cleanly_without_recording}, + ) + + with pytest.raises(RuntimeError, match="incomplete"): + analysis.simulate( + number_of_simulations=6, append=False, parallel=True, n_workers=2 + ) + + +def test_no_failure_path_waits_on_a_worker_without_a_bound(): + """No ``join`` in the parallel path is called without a timeout.""" + # An unbounded join anywhere in the parallel path puts back the hang that + # the bounded teardown exists to end, and it does so where it is hardest + # to notice: only when a worker is already stuck. + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + + unbounded = [ + node.lineno + for node in ast.walk(run_in_parallel) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "join" + and not node.args + and not node.keywords + ] + + assert not unbounded, f"join() with no timeout at lines {unbounded}" + + +def test_starting_a_worker_happens_inside_the_cleanup_scope(): + """A start that fails has to leave the workers before it accounted for.""" + # Structural for the same reason the unbounded-join check is: it only shows + # when a start has already failed, which a unit test cannot make a real + # process do. Outside the try, an interrupt there left children running. + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + guarded = [ + node + for handler in ast.walk(run_in_parallel) + if isinstance(handler, ast.Try) + for node in ast.walk(handler) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "start" + ] + + assert guarded, "Process.start() is not inside a try in __run_in_parallel" + + +def test_a_worker_is_recorded_only_once_it_has_started(): + """A process that never started cannot be joined or terminated.""" + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + lines = {"start": None, "append": None} + for node in ast.walk(run_in_parallel): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + if node.func.attr == "start": + lines["start"] = node.lineno + if node.func.attr == "append": + lines["append"] = node.lineno + + assert lines["start"] is not None and lines["append"] is not None + assert lines["start"] < lines["append"], "appended before it started" + + +def test_a_collector_cannot_take_over_the_simulation_index( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """A collector key called index is refused before the run touches a file.""" + # Measured before this was refused: every row was written with the + # collector's value, so the log said 999 twice for a two-simulation run + # and every check that reads an index was reading the wrong thing. + # Refused when the collector is handed over, which is before any file + # is opened, rather than at the end of a run that is already spoilt. + with pytest.raises(ValueError, match="index"): + MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"index": lambda flight: 999}, + ) + + +def test_a_collector_key_of_its_own_is_still_welcome( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """The control. Only the one reserved name is refused.""" + analysis = MonteCarlo( + filename=str(tmp_path / "ok"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"apogee_twice": lambda flight: 2 * flight.apogee}, + ) + + analysis.simulate(number_of_simulations=1, append=False) + + with open(analysis.output_file, "r", encoding="utf-8") as written: + row = json.loads(next(line for line in written if line.strip())) + # Not the value of the index: how a run numbers its simulations is + # settled elsewhere, and pinning it here would tie this to that. + assert "index" in row + assert "apogee_twice" in row diff --git a/tests/unit/simulation/test_monte_carlo_worker_exit.py b/tests/unit/simulation/test_monte_carlo_worker_exit.py new file mode 100644 index 000000000..5b44bd352 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_exit.py @@ -0,0 +1,78 @@ +import os +from types import SimpleNamespace + +import pytest + +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _refuse_a_worker_that_did_not_finish, +) + + +def _worker(exitcode): + return SimpleNamespace(exitcode=exitcode) + + +def test_workers_that_all_exited_cleanly_are_accepted(): + """A fleet that all exited zero raises nothing.""" + _refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)]) + + +def test_a_worker_killed_by_a_signal_is_refused(): + """A negative exit code names the worker and the signal that ended it.""" + with pytest.raises(RuntimeError, match=r"worker 1 with exit code -9"): + _refuse_a_worker_that_did_not_finish([_worker(0), _worker(-9)]) + + +def test_a_worker_that_exited_nonzero_is_refused(): + """A positive exit code is refused the same way a signal is.""" + with pytest.raises(RuntimeError, match=r"worker 0 with exit code 1"): + _refuse_a_worker_that_did_not_finish([_worker(1), _worker(0)]) + + +def test_every_unfinished_worker_is_named(): + """The message names each unfinished worker and leaves the clean ones out.""" + with pytest.raises(RuntimeError) as raised: + _refuse_a_worker_that_did_not_finish([_worker(-9), _worker(0), _worker(3)]) + + assert "worker 0" in str(raised.value) + assert "worker 2" in str(raised.value) + assert "worker 1" not in str(raised.value) + + +@pytest.mark.parametrize("exitcode", [None, -15, 2]) +def test_anything_but_a_clean_exit_is_refused(exitcode): + """``None`` counts as unfinished, not as finished.""" + with pytest.raises(RuntimeError): + _refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)]) + + +def _leave_without_recording(_flight): + """Ends the worker the way a kill or an out-of-memory exit does. + + ``os._exit`` rather than a signal, since ``SIGKILL`` is POSIX-only, and + reached through the data collector rather than a patched method, since a + ``spawn`` platform re-imports the module and would not see the patch. + """ + os._exit(1) + + +def test_a_worker_that_leaves_early_does_not_pass_as_a_finished_run( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """A worker leaving through ``os._exit`` makes ``simulate`` raise.""" + # The event the workers report through is set by their own handler, and + # this one leaves without running it, so the run used to return as though + # it had done every simulation it was asked for. + analysis = MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"leave": _leave_without_recording}, + ) + + with pytest.raises(RuntimeError, match="incomplete"): + analysis.simulate( + number_of_simulations=6, append=False, parallel=True, n_workers=2 + ) diff --git a/tests/unit/simulation/test_monte_carlo_worker_join.py b/tests/unit/simulation/test_monte_carlo_worker_join.py new file mode 100644 index 000000000..fc674ba75 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_join.py @@ -0,0 +1,184 @@ +import pytest + +from rocketpy.simulation.monte_carlo import _join_the_workers + + +class _Worker: + """A process that stops after a set number of polls, or never. + + ``never`` stands in for one blocked on a lock its dead sibling was holding, + which is the case an unbounded join waits out forever. + """ + + def __init__(self, exitcode=0, alive_for=0, never=False, ignores_terminate=False): + self.exitcode = None + self._final_exitcode = exitcode + self._alive_for = alive_for + self._never = never + self._ignores_terminate = ignores_terminate + self.joins = 0 + self.timeouts = [] + self.terminated = False + self.killed = False + + def is_alive(self): + return self.exitcode is None + + def join(self, timeout=None): + self.joins += 1 + self.timeouts.append(timeout) + # A real worker that never returns makes the caller hang, which is the + # bug. Reproducing that here would hang CI instead of reporting, so the + # stand-in gives up and says so. + assert self.joins < 200, "the join loop never stopped waiting" + if self._never or self.joins <= self._alive_for: + return + self.exitcode = self._final_exitcode + + def terminate(self): + self.terminated = True + if not self._ignores_terminate: + self.exitcode = -15 + + def kill(self): + self.killed = True + self.exitcode = -9 + + +class _Event: + def __init__(self, already_set=False): + self.was_set = already_set + + def is_set(self): + return self.was_set + + def set(self): + self.was_set = True + + +def test_a_run_where_every_worker_finishes_is_left_alone(): + """A healthy fleet is joined to completion and never terminated.""" + workers = [_Worker(alive_for=3), _Worker(alive_for=5)] + + _join_the_workers(workers, _Event(), grace_period=0) + + assert [worker.exitcode for worker in workers] == [0, 0] + assert not any(worker.terminated for worker in workers) + + +def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever(): + """One bad exit ends the wait for a sibling that never returns.""" + # The one that mattered. Without a bound this call never returns, so the + # parent never reaches the check that would have reported the failure. + died = _Worker(exitcode=-9, alive_for=1) + blocked = _Worker(never=True) + + _join_the_workers([died, blocked], _Event(), grace_period=0) + + assert blocked.terminated + + +def test_the_survivors_are_asked_before_they_are_ended(): + """The event is set before anything is terminated.""" + died = _Worker(exitcode=1, alive_for=1) + blocked = _Worker(never=True) + event = _Event() + + _join_the_workers([died, blocked], event, grace_period=0) + + assert event.was_set + + +def test_a_survivor_that_stops_on_its_own_is_not_terminated(): + """A worker that leaves during the grace period is left alone.""" + died = _Worker(exitcode=1, alive_for=1) + cooperative = _Worker(alive_for=2) + + _join_the_workers([died, cooperative], _Event(), grace_period=0) + + assert not cooperative.terminated + assert cooperative.exitcode == 0 + + +@pytest.mark.parametrize("exitcode", [-9, 1, 2]) +def test_any_bad_exit_starts_the_shutdown(exitcode): + """Signals and non-zero codes both start the shutdown.""" + died = _Worker(exitcode=exitcode, alive_for=1) + blocked = _Worker(never=True) + + _join_the_workers([died, blocked], _Event(), grace_period=0) + + assert blocked.terminated + + +def test_a_slow_run_is_never_bounded(): + """Elapsed time is not evidence: a slow fleet is polled, never stopped.""" + # Nothing here may act on how long a worker takes, only on it having died. + slow = _Worker(alive_for=50) + slower = _Worker(alive_for=80) + + _join_the_workers([slow, slower], _Event(), grace_period=0) + + assert not any(worker.terminated for worker in (slow, slower)) + assert slower.joins > 50 + + +def test_a_reported_failure_ends_the_wait(): + """A worker that reported leaves cleanly, so its exit status says nothing.""" + # Left alone this waits on the sibling for good, and the run is already + # short a simulation whichever way that goes. The sibling is asked first + # and gets the longer grace, since one that only read the event is working + # rather than blocked on a lock nobody owns. + reported = _Worker(exitcode=0, alive_for=1) + stuck = _Worker(never=True) + + _join_the_workers([reported, stuck], _Event(already_set=True), grace_period=0) + + assert stuck.terminated or stuck.killed + assert max(t for t in stuck.timeouts if t is not None) >= 1.0 + + +def test_a_sibling_that_finishes_inside_the_grace_is_not_ended(): + """Asked first, and a worker that leaves on its own is left to do it.""" + reported = _Worker(exitcode=0, alive_for=1) + finishing = _Worker(alive_for=1) + + _join_the_workers([reported, finishing], _Event(already_set=True), grace_period=0) + + assert not finishing.terminated + assert finishing.exitcode == 0 + + +def test_a_clean_run_is_not_stopped_by_an_event_nobody_set(): + """An unset event leaves a healthy run running.""" + first, second = _Worker(alive_for=2), _Worker(alive_for=3) + + _join_the_workers([first, second], _Event(), grace_period=0) + + assert not any(worker.terminated for worker in (first, second)) + + +def test_a_worker_that_ignores_terminate_is_killed(): + """Terminate can be ignored; the fleet still has to come down.""" + died = _Worker(exitcode=-9, alive_for=1) + stubborn = _Worker(never=True, ignores_terminate=True) + + _join_the_workers([died, stubborn], _Event(), grace_period=0) + + assert stubborn.terminated + assert stubborn.killed + + +def test_the_fleet_comes_down_on_one_deadline_not_one_each(): + """A stage gives the fleet one grace period between them, not each.""" + # Observed through what each worker is offered: with a deadline of its own + # every worker is given the whole grace, so a fleet of thirty takes thirty + # times as long to give up on. + died = _Worker(exitcode=-9, alive_for=1) + stuck = [_Worker(never=True) for _ in range(4)] + + _join_the_workers([died, *stuck], _Event(), grace_period=0.05) + + offered = [t for t in stuck[-1].timeouts if t is not None] + assert offered + assert min(offered) < 0.05 diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py new file mode 100644 index 000000000..9c5b9c896 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py @@ -0,0 +1,406 @@ +import json +import os +from contextlib import suppress +from types import SimpleNamespace + +import pytest + +from rocketpy.simulation import monte_carlo as mc_module +from rocketpy.simulation.monte_carlo import MonteCarlo + + +class _Mutex: + def __init__(self): + self.held = False + self.acquired = 0 + self.blocking = None + self.timeout = None + + def acquire(self, blocking=True, timeout=None): # the proxy's signature + self.acquired += 1 + self.blocking, self.timeout = blocking, timeout + self.held = True + return True + + def release(self): + self.held = False + + +class _MutexThatCannotBeTaken(_Mutex): + """A lock a dead holder never gave back: acquire waits out its bound.""" + + def acquire(self, blocking=True, timeout=None): + super().acquire(blocking, timeout) + self.held = False + return False + + +class _MutexThatBreaksOnAcquire(_Mutex): + """The manager is gone, so asking for the lock raises instead.""" + + def acquire(self, blocking=True, timeout=None): + super().acquire(blocking, timeout) + self.held = False + raise OSError("the manager is gone") + + +class _MutexThatBreaksOnRelease(_Mutex): + """The manager goes while the lock is held, so giving it back raises.""" + + def release(self): + raise OSError("the manager is gone") + + +class _ErrorEvent: + def __init__(self, refuse=False): + self.was_set = False + self.refuse = refuse + + def is_set(self): + return self.was_set + + def set(self): + if self.refuse: + raise OSError("the manager is gone") + self.was_set = True + + +def _raise_instead(message): + def refuse(*_args, **_kwargs): + raise OSError(message) + + return refuse + + +def _refusing_model(): + def refuse(_seed): + raise RuntimeError("the models would not reseed") + + return SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse) + + +def _a_worker(tmp_path, model, event=None): + study = MonteCarlo( + filename=str(tmp_path / "study"), + environment=model, + rocket=model, + flight=model, + ) + return study, event or _ErrorEvent() + + +def _run(study, monitor, error_event, mutex=None): + # Name-mangled: the producer is what each worker process runs, and nothing + # else in the suite calls it. + mutex = mutex or _Mutex() + study._MonteCarlo__sim_producer(42, monitor, mutex, error_event) + return mutex + + +def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys): + """A failure above the loop is reported against worker startup.""" + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + assert error_event.was_set + reported = capsys.readouterr().out + assert "worker startup" in reported + assert "the models would not reseed" in reported + + +def test_a_worker_that_fails_before_claiming_an_index_says_so(tmp_path, capsys): + """A failed claim is startup too, since no index was taken.""" + + def refuse(): + raise RuntimeError("the monitor would not hand out an index") + + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=refuse) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "worker startup" in capsys.readouterr().out + + +def test_a_worker_that_fails_inside_a_simulation_names_the_index( + tmp_path, capsys, monkeypatch +): + """A failure after a claim is reported against that index.""" + + # The control. An index is claimed and the simulation then fails, which is + # the path that already worked, so the report still has to name it. + def refuse(_self): + raise RuntimeError("the simulation would not run") + + monkeypatch.setattr( + MonteCarlo, "_MonteCarlo__run_single_simulation", refuse, raising=True + ) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 8) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "iteration 7" in capsys.readouterr().out + + +def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path): + """The error log gets a row even when no inputs were drawn.""" + # The caller is told to read the error file, and a traceback the worker + # printed is not there to be read once its output has been redirected. + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as recorded: + rows = [json.loads(line) for line in recorded if line.strip()] + assert len(rows) == 1 + assert rows[0]["index"] is None + assert rows[0]["stage"] == "worker startup" + assert "the models would not reseed" in rows[0]["error"] + + +@pytest.mark.parametrize("failing", ["_set_stochastic", "increment"]) +def test_a_worker_failure_never_raises_out_of_the_producer(tmp_path, failing): + """A reported failure leaves the producer without an exception.""" + + # The handler used to reach for names the loop had not bound yet, so the + # process died with UnboundLocalError and the parent waited forever. + def refuse(*_args): + raise RuntimeError("boom") + + model = SimpleNamespace( + last_rnd_dict={}, + _set_stochastic=refuse if failing == "_set_stochastic" else lambda _s: None, + ) + monitor = SimpleNamespace( + keep_simulating=lambda: True, + increment=refuse if failing == "increment" else (lambda: 1), + ) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + + +@pytest.mark.parametrize("breaking", ["error_file", "reprint", "event"]) +def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaking): + """The manager lock is released however the reporting goes.""" + # The mutex is the manager's, so a worker that ends while holding it leaves + # the next one waiting on a process that is gone, and the parent never + # reaches the join that would have noticed. + if breaking == "error_file": + monkeypatch.setattr( + mc_module, "_worker_failure_record", _raise_instead("no disk") + ) + if breaking == "reprint": + monkeypatch.setattr( + mc_module._SimMonitor, "reprint", _raise_instead("no stdout") + ) + event = _ErrorEvent(refuse=breaking == "event") + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model(), event) + mutex = _Mutex() + + # A worker that could not announce its failure re-raises on the way out, so + # that its exit code carries what the event could not. The lock still has + # to be back either way, which is what this is about. + with suppress(RuntimeError): + _run(study, monitor, error_event, mutex) + + assert mutex.acquired == 1 + assert not mutex.held + + +def test_a_reporting_failure_does_not_replace_the_simulation_failure( + tmp_path, monkeypatch, capsys +): + """An unwritable log does not hide what actually failed.""" + monkeypatch.setattr(mc_module, "_worker_failure_record", _raise_instead("no disk")) + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + assert not os.path.getsize(study.error_file) + + +def _committing_producer(monkeypatch): + """Make one simulation run start to finish without a real flight.""" + monkeypatch.setattr( + MonteCarlo, "_MonteCarlo__run_single_simulation", lambda self: None + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_inputs", + lambda self, index: json.dumps({"index": index, "committed": True}) + "\n", + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + lambda self, flight, index: json.dumps({"index": index}) + "\n", + ) + + +def _one_then_broken(): + calls = {"count": 0} + + def keep_simulating(): + calls["count"] += 1 + if calls["count"] == 1: + return True + raise RuntimeError("the monitor died between simulations") + + return SimpleNamespace( + keep_simulating=keep_simulating, + increment=lambda: 1, + print_update_status=lambda: None, + ) + + +def test_a_failure_between_simulations_is_not_blamed_on_the_last_one( + tmp_path, capsys, monkeypatch +): + """A failure after a committed row is not reported against it.""" + # Simulation 0 finishes and its row is committed. The next claim then + # fails, which is not simulation 0's doing and must not be recorded as it. + _committing_producer(monkeypatch) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + study, error_event = _a_worker(tmp_path, model) + + _run(study, _one_then_broken(), error_event) + + assert "worker startup" in capsys.readouterr().out + + +def test_a_committed_row_is_not_written_to_the_error_log_as_well(tmp_path, monkeypatch): + """A row that succeeded appears in one log, not in both.""" + _committing_producer(monkeypatch) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + study, error_event = _a_worker(tmp_path, model) + + _run(study, _one_then_broken(), error_event) + + with open(study.output_file, "r", encoding="utf-8") as written: + committed = [json.loads(line) for line in written if line.strip()] + with open(study.error_file, "r", encoding="utf-8") as recorded: + errored = [json.loads(line) for line in recorded if line.strip()] + + assert committed == [{"index": 0}] + assert all(row.get("committed") is None for row in errored) + + +def test_a_worker_that_cannot_announce_its_failure_does_not_exit_cleanly(tmp_path): + """With the event unreachable the producer raises, so the exit is not zero.""" + # The event is how a worker reaches the parent. With it unreachable, the + # only signal left is how the process ends, so it must not end well. + model = _refusing_model() + study, error_event = _a_worker(tmp_path, model, _ErrorEvent(refuse=True)) + + with pytest.raises(RuntimeError, match="the models would not reseed"): + _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event) + + +def test_a_worker_that_did_announce_its_failure_returns(tmp_path): + """With the event delivered the producer returns on purpose.""" + # The control. With the event delivered the parent already knows, so the + # producer returns and the process exits cleanly on purpose. + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event) + + assert error_event.was_set + + +@pytest.mark.parametrize( + "mutex_class", [_MutexThatCannotBeTaken, _MutexThatBreaksOnAcquire] +) +def test_a_lock_the_reporter_cannot_take_does_not_stop_it( + tmp_path, capsys, mutex_class +): + """A lock that times out or is gone still leaves the failure announced.""" + # Asking for it without a bound is how a worker whose sibling died holding + # the lock waits forever, with nothing recorded and no exit code to read. + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + mutex = mutex_class() + + _run(study, monitor, error_event, mutex) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + assert not mutex.held + assert mutex.timeout is not None # asked for with a bound + + +def test_a_lock_that_breaks_on_release_does_not_hide_the_failure(tmp_path, capsys): + """Giving the lock back can raise, and must not replace what failed.""" + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event, _MutexThatBreaksOnRelease()) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + + +def test_a_failure_after_the_inputs_were_drawn_still_records_why(tmp_path, monkeypatch): + """The error file is where the caller is sent, so it has to say what broke.""" + # Writing the input row on its own left no stage and no traceback there, + # for every failure past the point the inputs had been built. + _committing_producer(monkeypatch) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + _raise_instead("the outputs would not serialize"), + ) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + study, error_event = _a_worker( + tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) + ) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert len(rows) == 1 + assert "the outputs would not serialize" in rows[0]["error"] + assert rows[0]["stage"] == "iteration 0" + assert rows[0]["inputs"]["committed"] is True + + +def test_an_input_row_that_is_not_an_object_does_not_break_the_reporter( + tmp_path, monkeypatch +): + """Reading the row is best effort: the failure being reported comes first.""" + _committing_producer(monkeypatch) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_inputs", + lambda self, index: json.dumps([index]) + "\n", + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + _raise_instead("the outputs would not serialize"), + ) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + study, error_event = _a_worker( + tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) + ) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert "the outputs would not serialize" in rows[0]["error"] + assert "inputs" not in rows[0]