fix(bridge): camera process-race and unhandled JSON errors in two handlers
Found during a targeted code review, not from a user report: - _run_jpeg_loop()/_run_h264_loop() operated directly on the shared self._proc_jpeg/self._proc_h264 instance attributes in their cleanup, unlike _run_mjpeg_loop() (already fixed for exactly this) which uses a local `proc` reference. If a loop's task is cancelled - e.g. by CameraCache.reset() after the printer rotates its stream URL on reboot - while a new task has already started and assigned its own process to the shared attribute, the cancelled task's cleanup killed the NEWER process instead of its own, leaking its own ffmpeg child as an orphan. Applied the same local-variable + identity-check pattern already used by _run_mjpeg_loop. - handle_api_settings_post and handle_api_update_apply were the only two of ~84 handlers that called `await request.json()` without a try/except - every other handler follows the established pattern of returning a clean 400 for a malformed body. A trivial malformed request to either endpoint produced an unhandled 500 with a full traceback instead. New tests in tests/test_camera_process_race.py, tests/test_settings.py, and tests/test_update_check.py cover both.
This commit is contained in:
@ -713,7 +713,7 @@ class CameraCache:
|
||||
await asyncio.sleep(2.0)
|
||||
continue
|
||||
try:
|
||||
self._proc_jpeg = await asyncio.create_subprocess_exec(
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
_find_ffmpeg(), "-loglevel", "warning",
|
||||
*self._input_args(url), "-i", url,
|
||||
"-vf", "fps=2",
|
||||
@ -722,6 +722,7 @@ class CameraCache:
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
self._proc_jpeg = proc
|
||||
except Exception as e:
|
||||
log.warning(f"CameraCache: ffmpeg-jpeg start failed: {e}")
|
||||
await asyncio.sleep(3.0)
|
||||
@ -731,7 +732,7 @@ class CameraCache:
|
||||
rc = None
|
||||
try:
|
||||
while True:
|
||||
chunk = await self._proc_jpeg.stdout.read(self.TS_CHUNK)
|
||||
chunk = await proc.stdout.read(self.TS_CHUNK)
|
||||
if not chunk:
|
||||
break
|
||||
buf += chunk
|
||||
@ -751,26 +752,31 @@ class CameraCache:
|
||||
except Exception as e:
|
||||
log.debug(f"CameraCache: jpeg-loop unterbrochen: {e}")
|
||||
finally:
|
||||
# Kill + wait - otherwise the child process lingers as a zombie and
|
||||
# asyncio reports "Unknown child pid ..." on the next reaper tick.
|
||||
if self._proc_jpeg is not None:
|
||||
# NOTE: cleanup operates on the local `proc` reference, not on
|
||||
# self._proc_jpeg - see _run_mjpeg_loop's identical comment.
|
||||
# If this task got cancelled (e.g. by reset()), a new task may
|
||||
# already have started and assigned its own process to
|
||||
# self._proc_jpeg by the time we reach here; killing that
|
||||
# shared attribute instead of our own local proc would kill
|
||||
# the WRONG (newer) process and leak this one as an orphan.
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await proc.wait()
|
||||
except Exception:
|
||||
pass
|
||||
rc = proc.returncode
|
||||
if rc:
|
||||
try:
|
||||
self._proc_jpeg.kill()
|
||||
err = await proc.stderr.read(500)
|
||||
if err:
|
||||
log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await self._proc_jpeg.wait()
|
||||
except Exception:
|
||||
pass
|
||||
rc = self._proc_jpeg.returncode
|
||||
if rc:
|
||||
try:
|
||||
err = await self._proc_jpeg.stderr.read(500)
|
||||
if err:
|
||||
log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}")
|
||||
except Exception:
|
||||
pass
|
||||
self._proc_jpeg = None
|
||||
if self._proc_jpeg is proc:
|
||||
self._proc_jpeg = None
|
||||
if rc:
|
||||
self._fail_count_jpeg += 1
|
||||
delay = min(2.0 * (2 ** self._fail_count_jpeg), 300.0)
|
||||
@ -788,7 +794,7 @@ class CameraCache:
|
||||
await asyncio.sleep(2.0)
|
||||
continue
|
||||
try:
|
||||
self._proc_h264 = await asyncio.create_subprocess_exec(
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
_find_ffmpeg(), "-loglevel", "warning",
|
||||
*self._input_args(url), "-i", url,
|
||||
"-c:v", "copy", "-an",
|
||||
@ -796,6 +802,7 @@ class CameraCache:
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
self._proc_h264 = proc
|
||||
except Exception as e:
|
||||
log.warning(f"CameraCache: ffmpeg-h264 start failed: {e}")
|
||||
await asyncio.sleep(3.0)
|
||||
@ -804,7 +811,7 @@ class CameraCache:
|
||||
rc = None
|
||||
try:
|
||||
while True:
|
||||
chunk = await self._proc_h264.stdout.read(self.TS_CHUNK)
|
||||
chunk = await proc.stdout.read(self.TS_CHUNK)
|
||||
if not chunk:
|
||||
break
|
||||
# Fanout: non-blocking per subscriber; slow clients
|
||||
@ -822,24 +829,31 @@ class CameraCache:
|
||||
except Exception as e:
|
||||
log.debug(f"CameraCache: h264-loop unterbrochen: {e}")
|
||||
finally:
|
||||
if self._proc_h264 is not None:
|
||||
# NOTE: cleanup operates on the local `proc` reference, not on
|
||||
# self._proc_h264 - see _run_mjpeg_loop's identical comment.
|
||||
# If this task got cancelled (e.g. by reset()), a new task may
|
||||
# already have started and assigned its own process to
|
||||
# self._proc_h264 by the time we reach here; killing that
|
||||
# shared attribute instead of our own local proc would kill
|
||||
# the WRONG (newer) process and leak this one as an orphan.
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await proc.wait()
|
||||
except Exception:
|
||||
pass
|
||||
rc = proc.returncode
|
||||
if rc:
|
||||
try:
|
||||
self._proc_h264.kill()
|
||||
err = await proc.stderr.read(500)
|
||||
if err:
|
||||
log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await self._proc_h264.wait()
|
||||
except Exception:
|
||||
pass
|
||||
rc = self._proc_h264.returncode
|
||||
if rc:
|
||||
try:
|
||||
err = await self._proc_h264.stderr.read(500)
|
||||
if err:
|
||||
log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}")
|
||||
except Exception:
|
||||
pass
|
||||
self._proc_h264 = None
|
||||
if self._proc_h264 is proc:
|
||||
self._proc_h264 = None
|
||||
if rc:
|
||||
self._fail_count_h264 += 1
|
||||
delay = min(2.0 * (2 ** self._fail_count_h264), 300.0)
|
||||
@ -5093,7 +5107,10 @@ class KobraXBridge:
|
||||
|
||||
async def handle_api_settings_post(self, request):
|
||||
import configparser
|
||||
data = await request.json()
|
||||
try:
|
||||
data = await request.json()
|
||||
except Exception:
|
||||
return self._json_cors({"error": "invalid json"}, status=400)
|
||||
config_path = self._find_config_path()
|
||||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@ -5513,7 +5530,10 @@ class KobraXBridge:
|
||||
]
|
||||
|
||||
async def handle_api_update_apply(self, request):
|
||||
data = await request.json()
|
||||
try:
|
||||
data = await request.json()
|
||||
except Exception:
|
||||
return web.json_response({"error": "invalid json"}, status=400)
|
||||
new_tag = data.get("tag", "")
|
||||
if "nightly" in self._read_version():
|
||||
return web.json_response(
|
||||
|
||||
Reference in New Issue
Block a user