Change and read a job's steps under one lock.

The worker, the thread that reads the build's output and the window all change a job while the window draws it. A
step could be read with a new state and the value before it, the output's lines could change while they were
copied, and the change counter was incremented without a lock although the window learns from it that the job
ended. Every change and the counter now take the job's lock, and the job screen and the command line copy the steps
under it.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>

launcher/cli.py

@@ -31,12 +31,13 @@ def follow(job):
     seen = {}
     job.start()
     while job.result is None:
-        for s in job.steps:
-            key = (s.state, s.value if s.state != jobs.RUNNING else int(s.fraction * 20))
-            if seen.get(s.id) != key:
-                seen[s.id] = key
-                print(f'{s.label}: {s.value}' + (f' ({s.detail})' if s.detail and s.state == jobs.RUNNING else ''),
-                      flush=True)
+        with job.locked():   # each step as one change of the worker left it
+            steps = [(s.id, s.label, s.state, s.value, s.fraction, s.detail) for s in job.steps]
+        for sid, label, state, value, fraction, detail in steps:
+            key = (state, value if state != jobs.RUNNING else int(fraction * 20))
+            if seen.get(sid) != key:
+                seen[sid] = key
+                print(f'{label}: {value}' + (f' ({detail})' if detail and state == jobs.RUNNING else ''), flush=True)
         time.sleep(0.2)
     for s in job.steps:
         if seen.get(s.id, (None,))[0] != s.state:

launcher/jobs.py

@@ -147,7 +147,10 @@ class Job:
         self.finished = None
         self.after = None           # what the window does next ('restart', ...)
         self.note = ''              # a sentence for the home screen after the job (note_scan)
-        self._lock = threading.Lock()
+        # Held for every change of a step and of rev, and by a reader that wants the steps as one change left them
+        # (locked). The worker, the thread that reads the build's output and the window's thread (cancel) all change
+        # the job. Reentrant: progress() begins a step, and every change ends in _touch().
+        self._lock = threading.RLock()
 
     # ---- state the window reads
     def step(self, sid):
@@ -200,47 +203,68 @@ class Job:
     def elapsed(self):
         return (self.finished or time.time()) - self.started
 
+    def locked(self):
+        """The lock to read the steps under (`with job.locked():`): they are then as one change left them, a step's
+        state with the value and fraction that belong to it. Held only to copy what is read, never across a
+        wait."""
+        return self._lock
+
     # ---- changes, from the worker
     def _touch(self):
-        self.rev += 1
+        # Under the lock: "rev += 1" from two threads at once can write an older count back, and a reader that had
+        # seen the newer one then takes the next change for one it has drawn already. The job's last change is its
+        # result, so the window stayed on the job screen of a job that had ended.
+        with self._lock:
+            self.rev += 1
 
     def log(self, line):
-        self.lines.append(line)
-        self.ctx.log(line)
+        with self._lock:
+            self.lines.append(line)
+        self.ctx.log(line)   # outside the lock: it waits for the disk
         self._touch()
 
+    def last_lines(self, n):
+        """The last n lines of the job's output, copied under the lock: the thread that reads the build's output
+        appends while the window draws them, and a deque changed while it is read raises."""
+        with self._lock:
+            return list(self.lines)[-n:]
+
     def begin(self, sid, detail=''):
-        s = self.step(sid)
-        if s and s.state != RUNNING:
-            s.state, s.fraction, s.detail, s.t0, s.value = RUNNING, 0.0, detail, time.time(), '0%'
-            s.pace.clear()
-            self._touch()
+        with self._lock:
+            s = self.step(sid)
+            if s and s.state != RUNNING:
+                s.state, s.fraction, s.detail, s.t0, s.value = RUNNING, 0.0, detail, time.time(), '0%'
+                s.pace.clear()
+                self._touch()
 
     def progress(self, sid, fraction, detail=''):
-        s = self.step(sid)
-        if s:
-            if s.state == PENDING:
-                self.begin(sid)
-            s.fraction = max(0.0, min(1.0, fraction))
-            s.pace.append((time.time(), s.fraction))
-            s.value = f'{int(s.fraction * 100)}%'
-            if detail:
-                s.detail = detail
-            self._touch()
+        with self._lock:
+            s = self.step(sid)
+            if s:
+                if s.state == PENDING:
+                    self.begin(sid)
+                s.fraction = max(0.0, min(1.0, fraction))
+                s.pace.append((time.time(), s.fraction))
+                s.value = f'{int(s.fraction * 100)}%'
+                if detail:
+                    s.detail = detail
+                self._touch()
 
     def finish(self, sid, value=None):
-        s = self.step(sid)
-        if s:
-            s.seconds = time.time() - s.t0 if s.t0 else 0.0
-            s.state, s.fraction = DONE, 1.0
-            s.value = value or f'Done · {mmss(s.seconds)}'
-            self._touch()
+        with self._lock:
+            s = self.step(sid)
+            if s:
+                s.seconds = time.time() - s.t0 if s.t0 else 0.0
+                s.state, s.fraction = DONE, 1.0
+                s.value = value or f'Done · {mmss(s.seconds)}'
+                self._touch()
 
     def skip(self, sid, value='Unchanged'):
-        s = self.step(sid)
-        if s:
-            s.state, s.value = SKIPPED, value
-            self._touch()
+        with self._lock:
+            s = self.step(sid)
+            if s:
+                s.state, s.value = SKIPPED, value
+                self._touch()
 
     def cancelled(self):
         return self.cancel_flag.is_set()
@@ -313,10 +337,11 @@ class Job:
                 self.result, self.error = 'failed', str(e) or type(e).__name__
                 self.ctx.log(failure_line(e) if isinstance(e, PLAIN_FAILURES) else traceback.format_exc())
         finally:
-            for s in self.steps:
-                if s.state == RUNNING:
-                    s.state = FAILED if self.result == 'failed' else PENDING
-                    s.value = 'Stopped' if self.result == 'failed' else 'Cancelled'
+            with self._lock:
+                for s in self.steps:
+                    if s.state == RUNNING:
+                        s.state = FAILED if self.result == 'failed' else PENDING
+                        s.value = 'Stopped' if self.result == 'failed' else 'Cancelled'
             if marked:
                 try:
                     (self.ctx.layout.data / MARKER).unlink(missing_ok=True)

launcher/ui/app.py

@@ -2074,15 +2074,17 @@ class App:
     def rows_job(self):
         j = self.job
         rows = []
-        for st in j.steps:
-            if st.state == jobs.RUNNING:
-                rows.append(Row('progress', st.label, fraction=st.fraction, value=f'{int(st.fraction * 100)}%',
+        with j.locked():   # each step as one change of the worker left it, not a state with the value before it
+            steps = [(st.id, st.label, st.state, st.value, st.fraction) for st in j.steps]
+        for sid, label, state, value, fraction in steps:
+            if state == jobs.RUNNING:
+                rows.append(Row('progress', label, fraction=fraction, value=f'{int(fraction * 100)}%',
                                 focus_value=True, desc=None))
             else:
-                style = {jobs.DONE: 'normal', jobs.SKIPPED: 'dim', jobs.PENDING: 'dim', jobs.FAILED: 'salmon'}[st.state]
-                rows.append(Row('value', st.label, st.value, style, check=st.state == jobs.DONE, desc=None,
-                                focus_value=st.state == jobs.FAILED))
-            if st.id in ('romfs', 'verify') and st is not j.steps[-1]:
+                style = {jobs.DONE: 'normal', jobs.SKIPPED: 'dim', jobs.PENDING: 'dim', jobs.FAILED: 'salmon'}[state]
+                rows.append(Row('value', label, value, style, check=state == jobs.DONE, desc=None,
+                                focus_value=state == jobs.FAILED))
+            if sid in ('romfs', 'verify') and sid != steps[-1][0]:
                 rows.append(Row('divider'))
         if j.result is None:
             label = {'install': 'Cancel Installation', 'update': 'Cancel Update'}.get(j.kind, 'Cancel')
@@ -2105,7 +2107,7 @@ class App:
     def job_desc(self):
         j = self.job
         cur = j.current()
-        lines = [ln for ln in j.lines if ln.strip() and not ln.startswith('@@')][-6:]
+        lines = [ln for ln in j.last_lines(200) if ln.strip() and not ln.startswith('@@')][-6:]
         if j.result == 'failed':
             return {'title': 'The Build Stopped' if j.kind != 'update' or cur.id not in ('download', 'verify')
                     else 'The Update Stopped',
@@ -2165,7 +2167,7 @@ class App:
         j = self.job
         report = [f'NX2navi {self.layout.version} ({"checkout" if self.layout.dev else "release"}), {sys.platform}',
                   f'Job: {j.kind}, step: {j.current().label}', f'Error: {j.error}', '', 'Last lines:']
-        report += [ln for ln in list(j.lines)[-60:]]
+        report += j.last_lines(60)
         self.root.clipboard_clear()
         self.root.clipboard_append('\n'.join(report))
         self.message = 'Report copied.'

tests/launcher_test.py

@@ -3833,6 +3833,71 @@ class JobTest(Tmp):
         (game / 'romfs' / 'romfs.bin').unlink()   # the image is gone: only the dump brings it back
         self.assertEqual(jobs.InstallJob(ctx, self.tmp / 'g.nup', game).args()[0], '--nup')
 
+    def test_a_reader_sees_each_step_as_one_change_left_it(self):
+        """The worker, the build's reader thread and the window change a job while the window reads it. Under
+        job.locked() a step's state comes with the value and fraction that belong to it, every change counts in
+        rev (the window draws again when rev changed, and learns from it that the job ended), and the output's
+        last lines can be copied while lines arrive."""
+        ctx = self.make_ctx('ok')
+        self.addCleanup(ctx._file.close)
+        j = jobs.Job(ctx, ['lift', 'compile'])
+        stop = threading.Event()
+        wrong, errors = [], []
+
+        def work():
+            while not stop.is_set():
+                for sid in ('lift', 'compile'):
+                    j.begin(sid, 'starting')
+                    j.progress(sid, 0.5, 'half')
+                    j.finish(sid)
+                    j.log(f'working on {sid}')
+                    with j.locked():
+                        j.step(sid).state, j.step(sid).value, j.step(sid).fraction = jobs.PENDING, 'Waiting', 0.0
+
+        def read():
+            try:
+                while not stop.is_set():
+                    with j.locked():
+                        seen = [(s.state, s.value, s.fraction) for s in j.steps]
+                    for state, value, fraction in seen:
+                        ok = {jobs.PENDING: value == 'Waiting' and fraction == 0.0,
+                              jobs.RUNNING: value.endswith('%') and value == f'{int(fraction * 100)}%',
+                              jobs.DONE: value.startswith('Done') and fraction == 1.0}[state]
+                        if not ok:
+                            wrong.append((state, value, fraction))
+                    j.last_lines(6)
+            except Exception as e:   # noqa: BLE001 - whatever a reader meets is the failure
+                errors.append(e)
+
+        threads = [threading.Thread(target=work), threading.Thread(target=read), threading.Thread(target=read)]
+        for t in threads:
+            t.start()
+        time.sleep(0.5)
+        stop.set()
+        for t in threads:
+            t.join(30)
+        self.assertEqual((wrong[:3], errors), ([], []))
+
+        counted = jobs.Job(ctx, [])
+        touchers = [threading.Thread(target=lambda: [counted._touch() for _ in range(20000)]) for _ in range(4)]
+        for t in touchers:
+            t.start()
+        for t in touchers:
+            t.join(60)
+        self.assertEqual(counted.rev, 80000)                             # no change was lost
+        try:
+            from launcher.ui.app import App
+        except ImportError:
+            return
+        a = mock.MagicMock()                                             # the job screen's rows, from a real job
+        a.job = jobs.InstallJob(ctx, self.tmp / 'g.nup', self.tmp / 'game')
+        a.job.begin('tools')
+        a.job.progress('tools', 0.25)
+        rows = App.rows_job(a)
+        self.assertEqual((rows[0].kind, rows[0].value, rows[0].fraction), ('progress', '25%', 0.25))
+        self.assertEqual([r.kind for r in rows].count('divider'), 1)     # after the game data, as before
+        self.assertEqual(rows[-1].kind, 'action')                        # Cancel
+
 
 # ---------------------------------------------------------------- one launcher, one job