feat: fail a record that was running when its workspace ended

docs/developer/adr/0002-a-value-lives-while-something-keeps-it.md

@@ -64,8 +64,9 @@ History and the store survive.
 The values the workspace kept in memory do not.
 Each pending record is
 
-- cancelled, unless a persist request names it, or a record named by a persist request reads it, directly or through other pending records;
-- run, if every value it reads is a file, is written, or comes from a pending record that runs;
+- failed, if it had started: the workspace ended while it ran, perhaps because of it, as an out-of-memory kill does, so running it again could end the process again;
+- otherwise cancelled, unless a persist request names it, or a record named by a persist request reads it, directly or through other pending records;
+- otherwise run, if every value it reads is a file, is written, or comes from a pending record that runs;
 - failed otherwise, as the record of a freeze is, since its driver is gone.
 
 ## Alternatives considered
@@ -106,4 +107,4 @@ To be designed with the service (scipp/essapps#27):
 - What a finish records besides the status, such as the software environment.
 - When values are dropped from the store before history (system story H1), and by whom.
 - The file format of each output type, and the layout of the store in a proposal's area.
-- An upgrade while persisted records run. If the old instance finishes them, the new version waits, for hours after an overnight batch, since only one workspace writes a proposal's history. If the new instance runs them again, the time they had run is lost. Either way, the upgrade closes every workspace of the old instance.
+- An upgrade while persisted records run. An orderly stop of the old instance waits for its running records, so the new version waits, for hours after an overnight batch, since only one workspace writes a proposal's history. If the old instance is killed instead, its running records fail at the new instance's start, and a person or the application submits them again, so the time they had run is lost. Either way, the upgrade closes every workspace of the old instance.

docs/developer/prior-art/aiida.md

@@ -72,8 +72,8 @@ Whether the workspace retries before it fails a record, or marks the failure so
 **Checkpointed workflow state does not survive upgrades.** A WorkChain resumes after a daemon restart because its state is serialised between steps.
 The unreleased version says such checkpoints "cannot be continued after upgrading. Finish or terminate all active processes before upgrading" ([CHANGELOG L11-12][cl-head]).
 The FAQ says to restart the daemon after every code change ([faq.rst L29][faq]).
-Our workspace pickles nothing: after a restart it reruns persisted pending records from history ([ADR 0002](../adr/0002-a-value-lives-while-something-keeps-it.md)).
-On an upgrade with work in flight, AiiDA's answer is to drain first, the "old instance finishes them" option of ADR 0002, Open, and system story H2.
+Our workspace pickles nothing: after a restart it runs again, from history, the persisted pending records that had not started ([ADR 0002](../adr/0002-a-value-lives-while-something-keeps-it.md)).
+On an upgrade with work in flight, AiiDA's answer is to drain first, the orderly stop of ADR 0002, Open, and system story H2.
 Warm workers (scipp/essapps#27) hold imported code as AiiDA's daemon workers do, so a package upgrade must restart them.
 
 **A broker queue drifted from the database.** v1.0 made RabbitMQ mandatory for process control.

docs/developer/system-stories.md

@@ -110,7 +110,7 @@ Property: stored outputs can be dropped by proposal, label, or age. Every record
 ### H2. Upgrade with runs in flight
 
 Actor: DMSC. Goal: deploy a new version of the service while requests run and others wait on them.
-Property: every record pending at the upgrade with a persist request finishes without anyone submitting it again, and so do the records it reads. Whether the old instance finishes them before it stops, or the new one runs them again, is open ([ADR 0002](adr/0002-a-value-lives-while-something-keeps-it.md), Open); a record that reads an accumulator's driver finishes only if the old instance finishes it. The upgrade closes every workspace of the old instance, and pending work that nothing persists is cancelled. Records and persisted values written before the upgrade stay readable, also when the new version stores records in another schema. A workspace started on history that another workspace still writes is refused, so the old and the new version never write the same history.
+Property: every record pending at the upgrade with a persist request finishes without anyone submitting it again, and so do the records it reads. Closing a workspace waits for them, so the old instance finishes them before it stops; how long an upgrade may wait for that is open ([ADR 0002](adr/0002-a-value-lives-while-something-keeps-it.md), Open). A process that is to stop in this orderly way on SIGTERM has to close its workspaces when it receives the signal, and no code in the repository handles the signal. An instance killed instead leaves them pending: at the next start, a record that was running fails, since running it again could end the process again, and a record that had not started runs if what it reads is there. A record that reads an accumulator's driver finishes only if the old instance finishes it. The upgrade closes every workspace of the old instance, and pending work that nothing persists and that had not started is cancelled. Records and persisted values written before the upgrade stay readable, also when the new version stores records in another schema. A workspace started on history that another workspace still writes is refused, so the old and the new version never write the same history.
 
 ### H3. History ends when the work does
 

packages/essapps/docs/batch.md

@@ -263,7 +263,7 @@ $B reduce iofq.yaml runs.csv --label water-series
 
 `{'@latest': label, output: name}` is read when the command starts, as the output of the newest record under the label, of `member` if given, whoever submitted it. It is for a template, not for a cell of a table. The record of each row holds the ID that it was read as. A reference by ID is `{'@ref': output, record: ..., output: ...}`.
 
-One command holds the folder at a time. If a command was killed, the next one first finishes the rows that were running, so reduce again with `--remaining`. A workspace reads a persisted output by the type that its spec declares, so only one that runs the spec reads it. Nothing removes the files of the store.
+One command holds the folder at a time. If a command was killed, the next one first finishes the rows that had not started. A row that was running fails, since its workflow may have been what ended the process. Reduce again with `--remaining` to submit every row that has not completed, the failed ones included. A workspace reads a persisted output by the type that its spec declares, so only one that runs the spec reads it. Nothing removes the files of the store.
 
 ## The UI
 

packages/essapps/src/ess/apps/log.py

@@ -4,22 +4,25 @@
 The workspace's history: an append-only log of what ran, with which params,
 and what came of it, for one proposal, which each submission and opening names.
 
-There are four events: a submission, a record that finished, an accumulator
-that opened, and a push into one. A persist request is part of a submission or
-a freeze, and its write is the record's: the record finishes once the outputs
-are written. A freeze is a submission of one record of the plain request of the
-state frozen, as the accumulator and its number of pushes; the views list its
-request when it is read. The workspace's views, such as the records by ID, are
-built by applying the events in order, when they are appended and again when a
-workspace starts from an existing log.
-
-An event names only records and accumulators appended before it: a finish
-names its record, a push its accumulator, a record the records it reads, and
-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.
+There are five events: a submission, a record that started, a record that
+finished, an accumulator that opened, and a push into one. A record starts when
+a worker begins its work, once what it reads is there, so a record that started
+and did not finish was running when its workspace ended. A persist request is
+part of a submission or a freeze, and its write is the record's: the record
+finishes once the outputs are written. A freeze is a submission of one record of
+the plain request of the state frozen, as the accumulator and its number of
+pushes; the views list its request when it is read. The workspace's views,
+such as the records by ID, are built by applying the events in order, when they
+are appended and again when a workspace starts from an existing log.
+
+An event names only records and accumulators appended before it: a start and
+a finish name their record, a push its accumulator, a record the records it
+reads, and 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
+``{"format": 2}``, 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
@@ -83,6 +86,13 @@ class Submitted(BaseModel, frozen=True, extra='forbid'):
     records: tuple[NewRecord | NewStateRecord, ...]
 
 
+class Started(BaseModel, frozen=True, extra='forbid'):
+    """A record whose work began on a worker: what it reads was there."""
+
+    kind: Literal['started'] = 'started'
+    record: str
+
+
 class Finished(BaseModel, frozen=True, extra='forbid'):
     """
     A record that finished: ``failure`` says why it failed, or why the
@@ -129,15 +139,15 @@ class Pushed(BaseModel, frozen=True, extra='forbid'):
 
 
 Event = Annotated[
-    Submitted | Finished | Opened | Pushed,
+    Submitted | Started | Finished | Opened | Pushed,
     Field(discriminator='kind'),
 ]
 _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
+FORMAT = 2
 _HEADER = json.dumps({'format': FORMAT}).encode() + b'\n'
 
 

packages/essapps/src/ess/apps/protocol.py

@@ -470,11 +470,7 @@ class Store(Protocol):
     """
 
     def write(self, record: str, output: str, value: Any) -> None:
-        """
-        Write the value of an output; raises if it cannot. A write replaces
-        a value written before under the same names, as a record run again
-        after a restart writes its outputs again.
-        """
+        """Write the value of an output; raises if it cannot."""
         ...
 
     def read(self, record: str, output: str, tp: Any) -> Any:

packages/essapps/src/ess/apps/views.py

@@ -25,6 +25,7 @@
     NewStateRecord,
     Opened,
     Pushed,
+    Started,
     Submitted,
 )
 from .records import Changes, Record, Request, SpecId, Status
@@ -86,21 +87,23 @@ class Views:
     The records, labels, accumulators, and pushes that the events describe.
 
     ``finished`` holds the event that finished a record; a record without one
-    is pending. ``states`` holds the accumulator and number of pushes of each
-    record of a freeze. ``accumulators`` holds the event that opened each
-    accumulator, and ``pushes`` its pushes in order, each rows of its tables.
+    is pending. ``started`` names the records whose work began. ``states`` holds
+    the accumulator and number of pushes of each record of a freeze.
+    ``accumulators`` holds the event that opened each accumulator, and
+    ``pushes`` its pushes in order, each rows of its tables.
 
     ``persisted`` names the outputs each record persists; their write is the
     record's: pending while it is, done once it completes.
 
     ``feed`` holds the ID of a record each time an event submits or finishes
     it, in the order of the events. A position in it is a
-    cursor of :meth:`changes`. Opening an accumulator and pushing to it are not in
-    the feed.
+    cursor of :meth:`changes`. Starting a record, opening an accumulator, and
+    pushing to it are not in the feed.
     """
 
     def __init__(self) -> None:
         self.records = _Records(self)
+        self.started: set[str] = set()
         self.finished: dict[str, Finished] = {}
         self.labels: dict[str, list[str]] = {}
         self.states: dict[str, tuple[str, int]] = {}
@@ -115,6 +118,8 @@ def apply(self, event: Event) -> None:
                 for new in event.records:
                     self._add(new, event)
                     self.feed.append(new.id)
+            case Started():
+                self.started.add(event.record)
             case Finished():
                 self.feed.append(event.record)
                 self.finished[event.record] = event

packages/essapps/src/ess/apps/workspace.py

@@ -10,16 +10,17 @@
 its pieces fit.
 
 What the workspace knows of history is its log (see ``log.py``): every submission,
-finished record, opened accumulator, and push is appended as one event and then
-applied to its views (see ``views.py``). A change is checked before its event is
-appended, so the log holds only events that apply. A workspace given a log that
-already has events applies them first; a log holds the history of one proposal,
-so a log with an event of another proposal is refused. Nothing the earlier
-workspace kept survives, so only persisted work goes on (see
-:meth:`Workspace._restart`). The log is the workspace's only record of history, so
-if an append fails, for example on a full disk, the workspace no longer knows its
-own state and stops: every call waiting in it, and every later call, raises
-``RuntimeError`` from the append's error.
+started and finished record, opened accumulator, and push is appended as one
+event and then applied to its views (see ``views.py``). A change is checked
+before its event is appended, so the log holds only events that apply. A
+workspace given a log that already has events applies them first; a log holds
+the history of one proposal, so a log with an event of another proposal is
+refused. Nothing the earlier workspace kept survives, so only persisted work
+that had not started goes on (see :meth:`Workspace._restart`). The log is the
+workspace's only record of history, so if an append fails, for example on a
+full disk, the workspace no longer knows its own state and stops: every call
+waiting in it, and every later call, raises ``RuntimeError`` from the append's
+error.
 
 The workspace keeps the outputs of records, and its stages and accumulators. A
 client is a connection to the workspace: it keeps nothing, and every client reads
@@ -185,6 +186,7 @@
     NewStateRecord,
     Opened,
     Pushed,
+    Started,
     Submitted,
 )
 from .protocol import (
@@ -637,17 +639,21 @@ def _restart(self) -> None:
         lock held. Nothing the earlier workspace kept survived, so only persisted
         work goes on:
 
-        1. A pending record is cancelled unless it persists outputs, or a
-           pending record that persists outputs reads it, directly or through
-           other pending records.
-        2. Otherwise it runs if each value it reads is a file, is written,
+        1. A pending record fails if it started: the workspace ended while it
+           ran, perhaps because of it, as an out-of-memory kill does, so
+           running it again could end the process again. A person submits its
+           request again to run it.
+        2. Otherwise it is cancelled unless it persists outputs, or a pending
+           record that persists outputs reads it, directly or through other
+           pending records.
+        3. Otherwise it runs if each value it reads is a file, is written,
            or is an output of a pending record that runs, and fails if not.
            The record of a freeze fails, since the driver it reads is gone.
            A call through a stage runs as its plain request, since the stage
            is gone, so it reads the values of the stage's template too.
 
-        A record that had written its persisted outputs, but
-        not finished, runs again, and its writes replace what the store holds.
+        So a record that runs at a restart had not started, and has written
+        nothing.
         """
         views = self._views
         pending = [i for i in views.records if i not in views.finished]
@@ -658,7 +664,11 @@ def _restart(self) -> None:
                 needed.update(ref.record for ref in upstream)
         runs: set[str] = set()
         for record_id in pending:
-            if record_id not in needed:
+            if record_id in views.started:
+                self._finish(
+                    record_id, Status.FAILED, 'the workspace ended while it ran'
+                )
+            elif record_id not in needed:
                 self._finish(record_id, Status.CANCELLED, 'the workspace restarted')
             elif (reason := self._unreadable(record_id, runs)) is not None:
                 self._finish(
@@ -997,6 +1007,8 @@ def _run(self, record_id: str) -> None:
         """
         Run a record's workflow, holding the values it reads until it returns,
         and write the outputs persisted at its submission before it completes.
+        The record's start is logged before anything is read, so that a
+        restart fails a record that was running when the workspace ended.
 
         It holds them even once the record is cancelled, since the workflow
         still reads them, and lets go of them before the write. The record of
@@ -1006,6 +1018,7 @@ def _run(self, record_id: str) -> None:
         with self._changed:
             if record_id in self._views.finished:
                 return
+            self._append(Started(record=record_id))
             record = self._views.records[record_id]
             stage = self._stage_of.get(record_id)
             state = self._state_of.get(record_id)

packages/essapps/tests/batch/frames_test.py

@@ -22,7 +22,7 @@
     write_table,
 )
 from ess.apps.folder import open_folder
-from ess.apps.log import Finished, Log
+from ess.apps.log import Finished, Log, Started
 from ess.apps.testing import StagedFunction
 from ess.spec import Binding, WorkflowSpec, function_binding
 
@@ -329,12 +329,14 @@ def test_rows_pending_after_a_restart_finish_and_are_not_submitted_again(
     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
+    unstarted = [
+        ln
+        for ln, e in zip(lines, events, strict=True)
+        if not isinstance(e, Started | Finished)
     ]
-    assert len(unfinished) < len(lines)
-    # as if the process had been killed
-    log.write_bytes(b''.join([header, *unfinished]))
+    assert len(unstarted) < len(lines)
+    # as if the process had been killed before the records started
+    log.write_bytes(b''.join([header, *unstarted]))
 
     with open_folder(folder, folder_bindings) as second:
         result = reduce_table(