diff --git a/packages/essapps/src/ess/apps/client.py b/packages/essapps/src/ess/apps/client.py index 13509fa5..326016ff 100644 --- a/packages/essapps/src/ess/apps/client.py +++ b/packages/essapps/src/ess/apps/client.py @@ -514,14 +514,24 @@ def wait(self, records: list[Record]) -> list[Status]: The status of each record once all have finished; a failure is not raised. If the workspace closes meanwhile, raises ``RuntimeError``. """ - while True: - # The position comes first, so that no finish after it is missed. - position = self._position - statuses = [s.status for s in self._statuses(records)] - if Status.PENDING not in statuses: - return statuses + ids = [r.id for r in records] + # The position comes first, so that no finish after it is missed. + position = self._position + current = self._workspace.status(ids) + statuses = {i: s.status for i, s in zip(ids, current, strict=True)} + pending = {i for i, s in statuses.items() if s is Status.PENDING} + while pending: changes = self._workspace.changes(position, timeout=_WAIT) - self._position = changes.cursor + position = self._position = changes.cursor + # Only the records that changed are asked again, so that waiting for + # many records costs time in proportion to their changes. + if changed := [i for i in changes.records if i in pending]: + current = self._workspace.status(changed) + for i, s in zip(changed, current, strict=True): + statuses[i] = s.status + if s.status is not Status.PENDING: + pending.remove(i) + return [statuses[i] for i in ids] @_shaped def failure(self, records: list[Record]) -> list[str | None]: diff --git a/packages/essapps/src/ess/apps/folder.py b/packages/essapps/src/ess/apps/folder.py index ac5deedd..c032d2f9 100644 --- a/packages/essapps/src/ess/apps/folder.py +++ b/packages/essapps/src/ess/apps/folder.py @@ -8,7 +8,8 @@ files.csv the table of FolderFiles: file, run, kind, role, ... (written by hand, or started with scan_folder) - log.jsonl history (see ess.apps.log) + log.jsonl history: a header naming the format of the log and of + the store, then one event per line (see ess.apps.log) store/ persisted outputs, /.h5 So a command that runs and ends, and the next one, are clients of one workspace: @@ -62,7 +63,10 @@ def read(self, record: str, output: str, tp: Any) -> Any: path = self.path(record, output) if not path.is_file(): raise LookupError(f'the store holds no {output} of {record}') - return hdf5.read(tp, path) + try: + return hdf5.read(tp, path) + except ValueError as error: # written under other outputs of the spec + raise LookupError(str(error)) from None _OWN = ('files.csv', 'log.jsonl', 'store') diff --git a/packages/essapps/src/ess/apps/hdf5.py b/packages/essapps/src/ess/apps/hdf5.py index f9d0d69b..5ce1e033 100644 --- a/packages/essapps/src/ess/apps/hdf5.py +++ b/packages/essapps/src/ess/apps/hdf5.py @@ -20,7 +20,11 @@ A value that has no such form is refused with ``TypeError`` when it is written: a date, a tuple, ``None`` in a list, an object of any other class, and a ``DataGroup`` that holds one of these. A group is read back with the fields -that were written, not validated again, as an instance of the declared type. +that were written, not validated again, as an instance of the declared type: +a field the file lacks is ``None`` if ``None`` is a value of its type (an +optional type, ``Any``, or ``object``), else its default. +A file that lacks a required field, or holds a value an enum lacks, was written +under other outputs of the spec, and reading it raises ``ValueError``. """ from __future__ import annotations @@ -73,17 +77,26 @@ def _is(tp: Any, base: type) -> bool: return isinstance(tp, type) and issubclass(tp, base) +def _takes_none(tp: Any) -> bool: + """Whether ``None`` is a value of type ``tp``.""" + return tp is Any or tp is object or field_types.optional(tp)[1] + + def _typed(tp: Any, data: Any) -> Any: """``data`` of a file as the value of type ``tp`` that was written.""" tp, _ = field_types.optional(tp) if _is(tp, BaseModel): - # not validated again: the workflow returned a valid value - return tp.model_construct( - **{ - name: _typed(info.annotation, data[name]) if name in data else None - for name, info in tp.model_fields.items() - } - ) + fields: dict[str, Any] = {} + for name, info in tp.model_fields.items(): + if name in data: + fields[name] = _typed(info.annotation, data[name]) + elif _takes_none(info.annotation): + fields[name] = None # left out when written + elif info.is_required(): + raise ValueError(f'field {name!r} of {tp.__name__} is missing') + # Not validated again: the workflow returned a valid value, and a + # validator need not give the same value when it runs on its own result. + return tp.model_construct(**fields) if _is(tp, Enum): return tp(data) if get_origin(tp) is list: @@ -103,8 +116,17 @@ def write(value: Any, path: Path) -> None: def read(tp: Any, path: Path) -> Any: - """The value of type ``tp`` that :func:`write` wrote to ``path``.""" + """ + The value of type ``tp`` that :func:`write` wrote to ``path``. Raises + ``ValueError`` if the file holds no value of ``tp``, as the module says. + """ data = sc.io.load_hdf5(path) plain_type, _ = field_types.optional(tp) alone = plain_type in _ALONE_TYPES or _is(plain_type, Enum) - return _typed(tp, data[_ALONE] if alone else data) + try: + return _typed(tp, data[_ALONE] if alone else data) + except ValueError as error: + raise ValueError( + f'{path} holds no valid {getattr(tp, "__name__", tp)}: {error}. ' + "The spec's outputs changed since the value was written." + ) from None diff --git a/packages/essapps/src/ess/apps/log.py b/packages/essapps/src/ess/apps/log.py index 3bcfb200..fe7d5e6c 100644 --- a/packages/essapps/src/ess/apps/log.py +++ b/packages/essapps/src/ess/apps/log.py @@ -18,6 +18,10 @@ the record of a freeze its accumulator. Applying the events in order relies on this, and so does a restart, which walks the pending records backwards. +A log file starts with a header line that names its format, such as +``{"format": 1}``, and holds one event per line after it. A log reads only +files of its own format, :data:`FORMAT`, and leaves a file it refuses as it was. + Stages and accumulators are not history, and neither are output values or anything a binding computed; those have their own lifetime. A submission is one event, so a workspace that stops half-way through one has @@ -27,6 +31,7 @@ from __future__ import annotations import fcntl +import json import os from collections.abc import Iterator from datetime import datetime @@ -129,13 +134,39 @@ class Pushed(BaseModel, frozen=True, extra='forbid'): ] _event: TypeAdapter[Event] = TypeAdapter(Event) +# The format of a log file, named by its first line. Bumped on any change to +# the events or to the files of the store (ess.apps.hdf5), since stored values +# are found through the log. +FORMAT = 1 +_HEADER = json.dumps({'format': FORMAT}).encode() + b'\n' + + +def _format(line: bytes) -> Any: + """The format a header line names, or None if the line is no header.""" + try: + header = json.loads(line) + except ValueError: + return None + if isinstance(header, dict) and header.keys() == {'format'}: + return header['format'] + return None + def _events(path: Path, lines: list[bytes]) -> list[Event]: """ - The events of ``lines`` of the log file ``path``. Raises ``ValueError`` - naming the first line that is not an event of this format, such as one + The events of ``lines`` of the log file ``path``, the first line being the + header. Raises ``ValueError`` if the header is not the one of this format, + or naming the first line that is not an event of this format, such as one with a field an event lacks, which validating would drop silently. """ + if not lines: + return [] + # An integer only: JSON's true and 1.0 compare equal to 1 in Python. + if type(found := _format(lines[0])) is not int or found != FORMAT: + found_text = 'no format' if found is None else f'format {found}' + raise ValueError( + f'{path} is a log of {found_text}; this version reads format {FORMAT}' + ) def read(number: int, line: bytes) -> Event: try: @@ -146,7 +177,7 @@ def read(number: int, line: bytes) -> Event: f'{path}, line {number}: not an event of a log of this format: {error}' ) from None - return [read(number, line) for number, line in enumerate(lines, start=1)] + return [read(number, line) for number, line in enumerate(lines[1:], start=2)] def _complete_lines(data: bytes) -> list[bytes]: @@ -185,15 +216,21 @@ class Log: Events in the order they were appended. Without a path the log lives in memory. With a path it is a file of one - JSON event per line, and the log holds the file until :meth:`close`: a - second log on the same file, in this process or another, is refused, so - a file has one writer. The operating system lets go of the file when the - process ends, so a log opened after a crash takes it over. The events + JSON event per line, after a first line that names the format of the file, + ``{"format": FORMAT}``, written when the file is created. The header is + written into an empty file, and into one that holds only the beginning of + the header, cut short by a crash. A file whose first line is not the header + of :data:`FORMAT`, or that has no complete line and holds anything else, + raises ``ValueError`` and is left as it was. + The log holds the file until :meth:`close`: a second log on the same + file, in this process or another, is refused, so a file has one writer. + The operating system lets go of the file when the process ends, so a log + opened after a crash takes it over. The events already in the file are read first. A last line cut short, by a crash - while it was written, is dropped; any other line that is not an event of - this format raises ``ValueError``. A write that fails, for example on a - full disk, leaves the file as it was, so no later event follows a line cut - short. + while it was written, is dropped from the file once the lines before it + are read; any other line that is not an event of this format raises + ``ValueError``. A write that fails, for example on a full disk, leaves the + file as it was, so no later event follows a line cut short. """ def __init__(self, path: Path | None = None) -> None: @@ -203,9 +240,19 @@ def __init__(self, path: Path | None = None) -> None: self._file = file = _hold(path) try: file.seek(0) - lines = _complete_lines(file.readall()) - file.truncate(sum(map(len, lines))) + data = file.readall() + lines = _complete_lines(data) + if not lines: + # Empty, or a header cut short by a crash while it was written. + if not _HEADER.startswith(data): + raise ValueError( + f'{path} is not a log: it has no complete header line' + ) + file.truncate(0) + _append_line(file, _HEADER) + lines = [_HEADER] self._events = _events(path, lines) + file.truncate(sum(map(len, lines))) except BaseException: file.close() raise diff --git a/packages/essapps/src/ess/apps/protocol.py b/packages/essapps/src/ess/apps/protocol.py index 7db91cf3..ecfc7aab 100644 --- a/packages/essapps/src/ess/apps/protocol.py +++ b/packages/essapps/src/ess/apps/protocol.py @@ -482,7 +482,9 @@ def read(self, record: str, output: str, tp: Any) -> Any: The value of an output that was written. ``tp`` is the type that the spec declares for the output, by which a store whose format holds no types makes the value. Raises ``LookupError`` if the - store no longer holds it, for example once dropped to free disk space: + store no longer holds it, for example once dropped to free disk space, + or holds a value that does not fit ``tp``, written under other outputs + of the spec: a client's read of it then raises ``NotReadable``, as for a value that nothing keeps, and a request that reads it fails. """ diff --git a/packages/essapps/src/ess/apps/workspace.py b/packages/essapps/src/ess/apps/workspace.py index d0e0acf4..58bb6427 100644 --- a/packages/essapps/src/ess/apps/workspace.py +++ b/packages/essapps/src/ess/apps/workspace.py @@ -1160,7 +1160,9 @@ def _stored(self, ref: OutputRef) -> Any: try: value = self._store.read(ref.record, ref.output, tp) except LookupError as error: - raise NotReadable(f'{ref}: the store no longer holds it') from error + raise NotReadable( + f'{ref}: the store no longer holds it: {error}' + ) from error return _element(value, ref) def _let_go(self, keys: Iterable[_Read]) -> None: diff --git a/packages/essapps/tests/batch/frames_test.py b/packages/essapps/tests/batch/frames_test.py index 21a23f4b..8bf89687 100644 --- a/packages/essapps/tests/batch/frames_test.py +++ b/packages/essapps/tests/batch/frames_test.py @@ -326,14 +326,15 @@ def test_rows_pending_after_a_restart_finish_and_are_not_submitted_again( with open_folder(folder, folder_bindings) as first: done = reduce_table(first, template, table, label='scan', persist=True) log = folder / 'log.jsonl' - lines = log.read_bytes().splitlines(keepends=True) + header, *lines = log.read_bytes().splitlines(keepends=True) events = Log.read(log) assert len(lines) == len(events) unfinished = [ ln for ln, e in zip(lines, events, strict=True) if type(e) is not Finished ] assert len(unfinished) < len(lines) - log.write_bytes(b''.join(unfinished)) # as if the process had been killed + # as if the process had been killed + log.write_bytes(b''.join([header, *unfinished])) with open_folder(folder, folder_bindings) as second: result = reduce_table( diff --git a/packages/essapps/tests/folder_test.py b/packages/essapps/tests/folder_test.py index 9c36c2db..0f5f9fed 100644 --- a/packages/essapps/tests/folder_test.py +++ b/packages/essapps/tests/folder_test.py @@ -12,7 +12,7 @@ from ess.apps import Client, NotReadable, file from ess.apps.folder import FolderStore, open_folder, scan_folder from ess.apps.testing import read_file -from ess.spec import Binding, WorkflowSpec, function_binding +from ess.spec import Binding, Referenced, WorkflowSpec, function_binding from .batch import toys @@ -156,6 +156,75 @@ def test_a_request_of_a_second_session_references_a_persisted_output( assert second.output(total, 'total') == 4.0 +class Centre(pydantic.BaseModel): + x: float + + +class Centred(pydantic.BaseModel): + centre: Centre + + +class CentreWithZ(pydantic.BaseModel): + """``Centre`` after a field was added, the spec's version unchanged.""" + + x: float + z: float + + +class CentredWithZ(pydantic.BaseModel): + centre: CentreWithZ + + +class UseParams(pydantic.BaseModel): + centre: Referenced[CentreWithZ] + + +def _centre_spec(outputs: type[pydantic.BaseModel]) -> WorkflowSpec: + return WorkflowSpec( + name='centre', + version=1, + title='Centre', + description='A toy.', + params=NoParams, + outputs=outputs, + ) + + +USE = WorkflowSpec( + name='use', + version=1, + title='Use', + description='A toy.', + params=UseParams, + outputs=toys.Outputs, +) + + +def test_an_output_persisted_before_its_spec_gained_a_field_is_not_read( + folder: Path, +) -> None: + old = _centre_spec(Centred) + with open_folder( + folder, [function_binding(old, lambda _: Centred(centre=Centre(x=1.0)))] + ) as first: + record = first.submit(old, params={}, persist=True) + first.wait([record]) + + changed = _centre_spec(CentredWithZ) + bindings = [ + function_binding(changed, lambda _: None), + function_binding(USE, lambda params: toys.Outputs(total=params.centre.z)), + ] + message = "field 'z' of CentreWithZ is missing. The spec's outputs changed" + with open_folder(folder, bindings) as second: + (record,) = second.records(ids=[record.id]) + with pytest.raises(NotReadable, match=message): + second.output(record, 'centre') + use = second.submit(USE, params={'centre': record.ref('centre')}) + second.wait([use]) + assert message in second.failure(use) + + def test_a_folder_held_by_a_client_cannot_be_opened_again( open_client: Callable[[], Client], ) -> None: diff --git a/packages/essapps/tests/hdf5_test.py b/packages/essapps/tests/hdf5_test.py index 755decde..b3c19f40 100644 --- a/packages/essapps/tests/hdf5_test.py +++ b/packages/essapps/tests/hdf5_test.py @@ -153,6 +153,41 @@ def test_a_group_is_read_back_with_the_value_written_not_validated_again( assert hdf5.read(Scaled, path).value == 2000.0 +class FloatsAndMore(pydantic.BaseModel): + """``Floats`` after the outputs of its spec gained a field.""" + + a: float + b: float + c: float + + +def test_a_group_that_lacks_a_required_field_is_refused_naming_file_type_and_field( + tmp_path: Path, +) -> None: + path = tmp_path / 'out.h5' + hdf5.write(Floats(a=1.0, b=2.0), path) + + with pytest.raises(ValueError, match="field 'c' of FloatsAndMore") as error: + hdf5.read(FloatsAndMore, path) + assert str(path) in str(error.value) + assert 'outputs changed since the value was written' in str(error.value) + + +class Anything(pydantic.BaseModel): + a: float + any: Any + obj: object + + +def test_a_required_field_of_any_type_that_holds_none_is_read_back_as_none( + tmp_path: Path, +) -> None: + path = tmp_path / 'out.h5' + hdf5.write(Anything(a=1.0, any=None, obj=None), path) + + assert hdf5.read(Anything, path) == Anything(a=1.0, any=None, obj=None) + + def test_a_group_is_a_data_group_of_its_fields_without_those_that_are_none( tmp_path: Path, ) -> None: diff --git a/packages/essapps/tests/log_test.py b/packages/essapps/tests/log_test.py index 24044aa4..ab6be301 100644 --- a/packages/essapps/tests/log_test.py +++ b/packages/essapps/tests/log_test.py @@ -27,6 +27,7 @@ file, ) from ess.apps.log import ( + FORMAT, Event, Finished, Log, @@ -355,7 +356,7 @@ def test_a_reference_is_logged_in_its_json_form_and_read_back(tmp_path: Path) -> ] _write(tmp_path / 'log', *events) - submitted, pushed = (tmp_path / 'log').read_text().splitlines() + _, submitted, pushed = (tmp_path / 'log').read_text().splitlines() assert '"sample":{"@ref":"file","file":"uuid:run-1"}' in submitted assert '{"@ref":"output","record":"r1","output":"values","key":"0"}' in pushed assert Log.read(tmp_path / 'log') == events @@ -484,6 +485,82 @@ def test_a_write_that_fails_leaves_the_log_as_it_was( assert [type(e) for e in Log.read(path)] == [Submitted, Finished] +def test_a_new_log_file_starts_with_a_header_naming_its_format( + tmp_path: Path, +) -> None: + Log(tmp_path / 'log').close() + + assert (tmp_path / 'log').read_text().splitlines() == [ + json.dumps({'format': FORMAT}) + ] + + +def test_a_header_cut_short_is_completed_and_opens_an_empty_log( + tmp_path: Path, +) -> None: + path = tmp_path / 'log' + path.write_text(json.dumps({'format': FORMAT})[:5]) + + with Log(path) as log: + assert list(log) == [] + assert path.read_text() == json.dumps({'format': FORMAT}) + '\n' + + +def test_a_log_file_of_another_format_is_refused(tmp_path: Path) -> None: + path = tmp_path / 'log' + path.write_text(json.dumps({'format': FORMAT + 1}) + '\n') + + message = f'is a log of format {FORMAT + 1}; this version reads format {FORMAT}' + for read in (Log.read, Log): + with pytest.raises(ValueError, match=message): + read(path) + + +@pytest.mark.parametrize('found', [True, float(FORMAT)]) +def test_a_header_naming_the_format_other_than_as_an_integer_is_refused( + tmp_path: Path, found: Any +) -> None: + path = tmp_path / 'log' + path.write_text(json.dumps({'format': found}) + '\n') + + with pytest.raises(ValueError, match='this version reads format'): + Log(path) + + +def test_a_refused_log_file_with_a_last_line_cut_short_is_left_unchanged( + tmp_path: Path, +) -> None: + path = tmp_path / 'log' + content = json.dumps({'format': FORMAT + 1}).encode() + b'\n{"kind":"fin' + path.write_bytes(content) + + with pytest.raises(ValueError, match=f'is a log of format {FORMAT + 1}'): + Log(path) + assert path.read_bytes() == content + + +def test_a_file_without_a_newline_that_is_not_a_log_is_refused_unchanged( + tmp_path: Path, +) -> None: + path = tmp_path / 'log' + path.write_bytes(b'some text') + + with pytest.raises(ValueError, match='is not a log'): + Log(path) + assert path.read_bytes() == b'some text' + + +def test_a_log_file_without_a_header_is_refused(tmp_path: Path) -> None: + path = tmp_path / 'log' + finished = Finished(record='r1', status=Status.COMPLETED) + path.write_text(finished.model_dump_json() + '\n') + + message = f'is a log of no format; this version reads format {FORMAT}' + for read in (Log.read, Log): + with pytest.raises(ValueError, match=message): + read(path) + + def test_a_log_file_has_one_writer(tmp_path: Path) -> None: path = tmp_path / 'log' with Log(path), pytest.raises(RuntimeError, match='held by another log'): @@ -585,10 +662,10 @@ def test_a_line_with_a_field_an_event_lacks_is_refused_naming_the_line( ) -> None: path = tmp_path / 'log' finished = Finished(record='r0', status=Status.COMPLETED).model_dump_json() - path.write_text(f'{finished}\n{json.dumps(event)}\n') + path.write_text(f'{{"format": {FORMAT}}}\n{finished}\n{json.dumps(event)}\n') for read in (Log.read, Log): with pytest.raises( - ValueError, match=r'log, line 2: not an event of a log of this format' + ValueError, match=r'log, line 3: not an event of a log of this format' ): read(path)