Log format number, stored groups checked on read, and a linear Client.wait

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]:

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, <record>/<output>.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')

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

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

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.
         """

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:

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(

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: