From 744c0165e2a367ed853b562e304e9585903702ab Mon Sep 17 00:00:00 2001 From: VAIO73 <50487331+Vaso73@users.noreply.github.com> Date: Mon, 26 Oct 2026 07:57:20 +0000 Subject: [PATCH] fix: keep update viewers responsive under input backpressure --- AppImage/components/script-terminal-modal.tsx | 34 ++++ AppImage/scripts/flask_terminal_routes.py | 7 +- AppImage/scripts/lxc_update_jobs.py | 49 +++++- tests/test_lxc_update_jobs.cjs | 13 +- tests/test_lxc_update_jobs.py | 155 ++++++++++++++++++ 5 files changed, 250 insertions(+), 8 deletions(-) diff --git a/AppImage/components/script-terminal-modal.tsx b/AppImage/components/script-terminal-modal.tsx index aa3b8ec1..c1703c33 100644 --- a/AppImage/components/script-terminal-modal.tsx +++ b/AppImage/components/script-terminal-modal.tsx @@ -222,18 +222,30 @@ export function ScriptTerminalModal({ try { const msg = JSON.parse(event.data) if (msg.type === "job_output") { + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) if (msg.truncated) termRef.current?.writeln(t("vmLxc.updates.backgroundOutputTrimmed")) termRef.current?.write(msg.text) outputOffsetRef.current = msg.offset return } if (msg.type === "job_unavailable") { + setCurrentInteraction(null) + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) termRef.current?.writeln(`\x1b[31m${msg.message}\x1b[0m`) completionReceivedRef.current = true setIsComplete(true) ws.close() return } + if (backgroundUpdate && msg.type === "web_interaction" && msg.interaction === null) { + setCurrentInteraction(null) + setInteractionInput("") + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) + return + } if (msg.type === "web_interaction" && msg.interaction) { setIsWaitingNextInteraction(false) if (waitingTimeoutRef.current) { @@ -254,6 +266,11 @@ export function ScriptTerminalModal({ return } if (msg.type === "script_complete") { + if (backgroundUpdate) { + setCurrentInteraction(null) + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) + } const exitCode = Number(msg.exit_code ?? 1) completionReceivedRef.current = true setIsComplete(true) @@ -462,12 +479,17 @@ export function ScriptTerminalModal({ try { const msg = JSON.parse(event.data) if (msg.type === "job_output") { + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) if (msg.truncated) term.writeln(t("vmLxc.updates.backgroundOutputTrimmed")) term.write(msg.text) outputOffsetRef.current = msg.offset return } if (msg.type === "job_unavailable") { + setCurrentInteraction(null) + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) term.writeln(`\x1b[31m${msg.message}\x1b[0m`) completionReceivedRef.current = true setIsComplete(true) @@ -475,6 +497,13 @@ export function ScriptTerminalModal({ return } + if (backgroundUpdate && msg.type === "web_interaction" && msg.interaction === null) { + setCurrentInteraction(null) + setInteractionInput("") + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) + return + } if (msg.type === "web_interaction" && msg.interaction) { setIsWaitingNextInteraction(false) if (waitingTimeoutRef.current) { @@ -496,6 +525,11 @@ export function ScriptTerminalModal({ return } if (msg.type === "script_complete") { + if (backgroundUpdate) { + setCurrentInteraction(null) + setIsWaitingNextInteraction(false) + if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current) + } const exitCode = Number(msg.exit_code ?? 1) completionReceivedRef.current = true setIsComplete(true) diff --git a/AppImage/scripts/flask_terminal_routes.py b/AppImage/scripts/flask_terminal_routes.py index 228c482c..45d38270 100644 --- a/AppImage/scripts/flask_terminal_routes.py +++ b/AppImage/scripts/flask_terminal_routes.py @@ -171,9 +171,10 @@ def lxc_update_viewer(ws, run_id): offset = view['offset'] if view['revision'] != revision: revision = view['revision'] - if view['interaction']: - ws.send(json.dumps({'type': 'web_interaction', - 'interaction': view['interaction']})) + # A second viewer must also dismiss a question already + # answered elsewhere, or cleared when the job ends. + ws.send(json.dumps({'type': 'web_interaction', + 'interaction': view['interaction']})) if view['state'] == 'complete': ws.send(json.dumps({'type': 'script_complete', 'exit_code': view['exit_code']})) break diff --git a/AppImage/scripts/lxc_update_jobs.py b/AppImage/scripts/lxc_update_jobs.py index d3bd74a4..8664559e 100644 --- a/AppImage/scripts/lxc_update_jobs.py +++ b/AppImage/scripts/lxc_update_jobs.py @@ -40,6 +40,7 @@ MAX_RUNTIME = 4 * 3600 STOP_GRACE = 300 RESULT_TTL = 6 * 3600 MAX_OUTPUT = 256 * 1024 +MAX_INPUT = 64 * 1024 MAX_JOBS = 32 _lock = threading.RLock() _jobs = {} @@ -91,11 +92,15 @@ class UpdateJob: self.process = None self.log_path = None self._interaction_buffer = "" + self._input = bytearray() def launch(self): master, slave = pty.openpty() self.master = master try: + # A viewer's paste must never block the worker's output, timeout + # or cancellation when the updater does not read its stdin. + os.set_blocking(master, False) fd, self.log_path = tempfile.mkstemp(prefix="proxmenux_update_", suffix=".log") os.close(fd) fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 30, 120, 0, 0)) @@ -158,8 +163,9 @@ class UpdateJob: elif kind in ("input", "inputbox"): value["default"] = base64.b64decode(parts[4]).decode() with self.lock: - self.interaction = value - self.revision += 1 + if not self.stop_reason: + self.interaction = value + self.revision += 1 except (ValueError, UnicodeError, IndexError): continue @@ -168,6 +174,9 @@ class UpdateJob: if self.process.poll() is not None or self.stop_reason: return self.stop_reason, self.stop_at = reason, time.monotonic() + self._input.clear() + self.interaction = None + self.revision += 1 try: os.killpg(self.process.pid, signal.SIGTERM) except ProcessLookupError: @@ -176,6 +185,25 @@ class UpdateJob: def cancel(self): self._signal("cancelled") + def _write_input(self): + # The descriptor is nonblocking, so even a full PTY cannot hold this + # lock. Retain unwritten bytes in order until the child can read them. + with self.lock: + if not self._input or self.stop_reason: + return + try: + written = os.write(self.master, self._input[:16384]) + except BlockingIOError: + return + except OSError as exc: + if exc.errno != errno.EIO: + raise + # The child can close its terminal and keep working. Losing + # stdin must not turn that update into a viewer failure. + self._input.clear() + return + del self._input[:written] + def _run(self): decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") eof, exited_at = False, None @@ -196,10 +224,18 @@ class UpdateJob: except ProcessLookupError: pass self._interactions(interactions) - readable, _, _ = select.select([] if eof else [self.master], [], [], 0.05) + with self.lock: + pending_input = bool(self._input) and running and not self.stop_reason + readable, writable, _ = select.select( + [] if eof else [self.master], + [self.master] if pending_input else [], [], 0.05) + if writable: + self._write_input() if readable: try: data = os.read(self.master, 16384) + except BlockingIOError: + continue except OSError as exc: if exc.errno != errno.EIO: raise @@ -232,6 +268,7 @@ class UpdateJob: self.exit_code = code self.state = "finalizing" self.interaction = None + self._input.clear() self.revision += 1 try: self.callback(self.script, self.params, code, time.monotonic() - self.started) @@ -266,7 +303,11 @@ class UpdateJob: if message == "\x03": self.cancel() else: - os.write(self.master, message.encode()) + data = message.encode() + if len(self._input) + len(data) > MAX_INPUT: + raise ValueError("Terminal input buffer is full") + if not self.stop_reason: + self._input.extend(data) elif message.get("type") == "cancel": self.cancel() elif message.get("type") == "resize": diff --git a/tests/test_lxc_update_jobs.cjs b/tests/test_lxc_update_jobs.cjs index 84524056..d15449ac 100644 --- a/tests/test_lxc_update_jobs.cjs +++ b/tests/test_lxc_update_jobs.cjs @@ -32,7 +32,8 @@ function fixture(backgroundUpdate) { reconnectAttemptsRef: {current: 0}, reconnectTimeoutRef: {current: null}, onCompleteRef: {current: code => finished.push(code)}, setIsComplete: value => context.isComplete = value, setConnectionStatus() {}, - setIsWaitingNextInteraction() {}, setCurrentInteraction: value => context.interaction = value, + setIsWaitingNextInteraction: value => context.waiting = value, + setCurrentInteraction: value => context.interaction = value, setInteractionInput() {}, clearTimeout() {}, clearInterval() {}, setTimeout: fn => { fn(); return 1 }, attemptReconnect: () => reconnects.push(true), t: key => key, getCompletionMessage: code => 'Result ' + code, @@ -52,8 +53,18 @@ for (const handler of handlers.onmessage) { assert.equal(f.context.outputOffsetRef.current, 27) message({data: JSON.stringify({type: 'web_interaction', interaction: {id:'pending', type:'yesno'}})}) assert.equal(f.context.interaction.id, 'pending', 'replayed interaction remains answerable') + f.context.waiting = true + message({data: JSON.stringify({type: 'web_interaction', interaction: null})}) + assert.equal(f.context.interaction, null, 'answer in another viewer dismisses the stale question') + assert.equal(f.context.waiting, false) + f.context.waiting = true + message({data: JSON.stringify({type: 'job_output', text: 'working', offset: 34})}) + assert.equal(f.context.waiting, false, 'ordinary progress clears the post-answer spinner') + f.context.interaction = {id: 'stale'}; f.context.waiting = true message({data: JSON.stringify({type: 'script_complete', exit_code: 3})}) assert.deepEqual(f.finished, [3]) + assert.equal(f.context.interaction, null, 'a completed job cannot retain an active question') + assert.equal(f.context.waiting, false, 'a completed job cannot retain a processing overlay') message({data: JSON.stringify({type: 'script_complete', exit_code: 0})}) assert.deepEqual(f.finished, [3], 'replayed completion cannot finalize twice') const missing = fixture({runId: 'missing'}) diff --git a/tests/test_lxc_update_jobs.py b/tests/test_lxc_update_jobs.py index 26a13473..4f523424 100644 --- a/tests/test_lxc_update_jobs.py +++ b/tests/test_lxc_update_jobs.py @@ -231,6 +231,140 @@ class JobTests(unittest.TestCase): files.assert_not_called() self.finish(job) + def test_large_paste_cannot_block_output_cancel_or_timeout(self): + self.script.write_text("stty raw -echo\necho ready\nwhile :; do echo progress; sleep .02; done\n") + for run, cancel in [("cancel-paste", True), ("timeout-paste", False)]: + with self.subTest(run=run), patch.object(jobs, "MAX_RUNTIME", .5), \ + patch.object(jobs, "STOP_GRACE", .1): + job = self.start(run) + wait_for(lambda: "ready" in job.read(0)["text"]) + before = job.read(0)["offset"] + job.input("x" * 16384) + job.input("y" * 16384) + wait_for(lambda: job.read(0)["offset"] > before) + if cancel: + job.cancel() + wait_for(lambda: job.state == "complete") + self.assertEqual(job.exit_code, 130 if cancel else 124) + self.assertIsNotNone(job.process.poll()) + self.assertFalse(job._input) + self.assertEqual(self.callback.call_count, 2) + + def test_queued_paste_reaches_a_slow_reader_in_order_without_losing_bytes(self): + payload = ("Ž🙂abc" * 3500).encode() + received = self.folder / "received" + reader = ( + "import os,pathlib; data=bytearray(); " + f"size={len(payload)}; " + "exec('while len(data)