Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
36764885ed | ||
|
|
5a82560cf2 | ||
|
|
7b1c6f6772 | ||
|
|
c025e7af95 | ||
|
|
3bad311e35 | ||
|
|
fdc4fe0a3e | ||
|
|
5df2308bdb | ||
|
|
f4c44ce1aa | ||
|
|
49232f11cc | ||
|
|
1258eff42d |
+3
-1
@@ -40,7 +40,9 @@ async def watch(req, ws):
|
|||||||
if sso.paskia_enabled():
|
if sso.paskia_enabled():
|
||||||
# SSO auth: call validation to get user info (don't enforce auth in public mode)
|
# SSO auth: call validation to get user info (don't enforce auth in public mode)
|
||||||
try:
|
try:
|
||||||
await sso.validate_sso_request(req)
|
# WebSocket cannot forward Set-Cookie, so ask the auth backend not to
|
||||||
|
# renew the session here; renewal happens on the HTTP side instead.
|
||||||
|
await sso.validate_sso_request(req, renew=False)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.debug("watch SSO validation failed: %s", e)
|
logger.debug("watch SSO validation failed: %s", e)
|
||||||
if sso_user := getattr(req.ctx, "sso_user", None):
|
if sso_user := getattr(req.ctx, "sso_user", None):
|
||||||
|
|||||||
+1
-1
@@ -86,7 +86,7 @@ async def log_access(req, res):
|
|||||||
path = f"{path}?{qs}"
|
path = f"{path}?{qs}"
|
||||||
extra = getattr(req.ctx, "log_extra", None)
|
extra = getattr(req.ctx, "log_extra", None)
|
||||||
line = format_access_log(
|
line = format_access_log(
|
||||||
client, res.status, req.method, host, path, duration_ms, extra=extra
|
client, res.status, req.method, host, path, duration_ms=duration_ms, extra=extra
|
||||||
)
|
)
|
||||||
access_logger.info(line)
|
access_logger.info(line)
|
||||||
return res
|
return res
|
||||||
|
|||||||
+154
-119
@@ -77,6 +77,9 @@ _preview_cache = PreviewCache(capacity=500)
|
|||||||
|
|
||||||
PREVIEW_TIMEOUT = 10.0 # seconds until preview subprocess is killed
|
PREVIEW_TIMEOUT = 10.0 # seconds until preview subprocess is killed
|
||||||
PREVIEW_WORKERS = max(2, min(8, cpu_count()))
|
PREVIEW_WORKERS = max(2, min(8, cpu_count()))
|
||||||
|
WORKER_KILL_GRACE = 5.0 # max seconds to wait for a killed worker to be reaped
|
||||||
|
WORKER_RESPAWN_DELAY = 1.0 # initial delay before retrying a failed worker spawn
|
||||||
|
WORKER_RESPAWN_DELAY_MAX = 30.0
|
||||||
_active_procs: set[asyncio.subprocess.Process] = set()
|
_active_procs: set[asyncio.subprocess.Process] = set()
|
||||||
_preview_pool = None
|
_preview_pool = None
|
||||||
_preview_pool_lock = asyncio.Lock()
|
_preview_pool_lock = asyncio.Lock()
|
||||||
@@ -144,11 +147,25 @@ class _PreviewWorker:
|
|||||||
return payload or None, resp
|
return payload or None, resp
|
||||||
|
|
||||||
async def kill(self) -> None:
|
async def kill(self) -> None:
|
||||||
if self.proc.returncode is None:
|
try:
|
||||||
with contextlib.suppress(ProcessLookupError):
|
if self.proc.returncode is None:
|
||||||
self.proc.kill()
|
# Safe to hard-kill: the worker is stateless per request.
|
||||||
await self.proc.wait()
|
# proc.wait() must not be awaited unaided: if a pipe
|
||||||
_active_procs.discard(self.proc)
|
# transport is flow-control paused (e.g. an undrained stderr
|
||||||
|
# pipe), asyncio may never resolve wait() even after SIGKILL,
|
||||||
|
# which would permanently wedge the calling dispatcher.
|
||||||
|
with contextlib.suppress(ProcessLookupError):
|
||||||
|
self.proc.kill()
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(self.proc.wait(), timeout=WORKER_KILL_GRACE)
|
||||||
|
except TimeoutError:
|
||||||
|
logger.error(
|
||||||
|
"Preview worker pid=%s not reaped within %ds of kill",
|
||||||
|
self.proc.pid,
|
||||||
|
int(WORKER_KILL_GRACE),
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
_active_procs.discard(self.proc)
|
||||||
|
|
||||||
|
|
||||||
class _PreviewWorkerPool:
|
class _PreviewWorkerPool:
|
||||||
@@ -163,23 +180,19 @@ class _PreviewWorkerPool:
|
|||||||
self._seq = 0
|
self._seq = 0
|
||||||
self._closed = False
|
self._closed = False
|
||||||
|
|
||||||
async def _read_startup_stderr(self, proc: asyncio.subprocess.Process) -> str:
|
|
||||||
if proc.stderr is None:
|
|
||||||
return ""
|
|
||||||
with contextlib.suppress(TimeoutError):
|
|
||||||
data = await asyncio.wait_for(proc.stderr.read(), timeout=0.5)
|
|
||||||
return data.decode(errors="replace").strip()
|
|
||||||
return ""
|
|
||||||
|
|
||||||
async def _spawn_worker(self) -> _PreviewWorker:
|
async def _spawn_worker(self) -> _PreviewWorker:
|
||||||
|
# stderr is inherited, not piped: a piped stderr that nobody drains
|
||||||
|
# eventually fills its OS buffer, blocking the worker mid-request,
|
||||||
|
# and its flow-control-paused transport makes proc.wait() hang even
|
||||||
|
# after kill() — together this used to permanently wedge the pool.
|
||||||
|
# Inheriting sends worker diagnostics straight to the server log.
|
||||||
proc = await asyncio.create_subprocess_exec(
|
proc = await asyncio.create_subprocess_exec(
|
||||||
sys.executable,
|
sys.executable,
|
||||||
"-m",
|
"-m",
|
||||||
"cista.preview_worker",
|
"cista.preview_worker",
|
||||||
stdin=asyncio.subprocess.PIPE,
|
stdin=asyncio.subprocess.PIPE,
|
||||||
stdout=asyncio.subprocess.PIPE,
|
stdout=asyncio.subprocess.PIPE,
|
||||||
stderr=asyncio.subprocess.PIPE,
|
stderr=None,
|
||||||
start_new_session=True,
|
|
||||||
)
|
)
|
||||||
_active_procs.add(proc)
|
_active_procs.add(proc)
|
||||||
try:
|
try:
|
||||||
@@ -189,21 +202,14 @@ class _PreviewWorkerPool:
|
|||||||
proc.kill()
|
proc.kill()
|
||||||
with contextlib.suppress(Exception):
|
with contextlib.suppress(Exception):
|
||||||
await proc.wait()
|
await proc.wait()
|
||||||
stderr = await self._read_startup_stderr(proc)
|
raise WorkerProtocolError(
|
||||||
if stderr:
|
"preview worker failed to become ready"
|
||||||
raise WorkerProtocolError(
|
" (worker stderr goes to the server log)"
|
||||||
"preview worker failed to become ready: " + stderr.splitlines()[-1]
|
) from err
|
||||||
) from err
|
|
||||||
raise WorkerProtocolError("preview worker failed to become ready") from err
|
|
||||||
except asyncio.IncompleteReadError as err:
|
except asyncio.IncompleteReadError as err:
|
||||||
stderr = await self._read_startup_stderr(proc)
|
|
||||||
if stderr:
|
|
||||||
raise WorkerProtocolError(
|
|
||||||
"preview worker exited before signalling readiness: "
|
|
||||||
+ stderr.splitlines()[-1]
|
|
||||||
) from err
|
|
||||||
raise WorkerProtocolError(
|
raise WorkerProtocolError(
|
||||||
"preview worker exited before signalling readiness"
|
"preview worker exited before signalling readiness"
|
||||||
|
" (worker stderr goes to the server log)"
|
||||||
) from err
|
) from err
|
||||||
if ready != b"\x01":
|
if ready != b"\x01":
|
||||||
raise WorkerProtocolError(f"preview worker ready signal invalid: {ready!r}")
|
raise WorkerProtocolError(f"preview worker ready signal invalid: {ready!r}")
|
||||||
@@ -216,106 +222,135 @@ class _PreviewWorkerPool:
|
|||||||
|
|
||||||
async def _replace_worker(self, worker: _PreviewWorker) -> None:
|
async def _replace_worker(self, worker: _PreviewWorker) -> None:
|
||||||
self._workers.discard(worker)
|
self._workers.discard(worker)
|
||||||
await worker.kill()
|
|
||||||
if self._closed:
|
|
||||||
return
|
|
||||||
try:
|
try:
|
||||||
await self._add_worker()
|
await worker.kill()
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Failed to replace preview worker")
|
logger.exception("Failed to kill preview worker pid=%s", worker.proc.pid)
|
||||||
|
# Keep retrying until a replacement is up: a pool that silently
|
||||||
async def _dispatch_loop(self) -> None:
|
# shrinks degrades all preview traffic to timeouts.
|
||||||
while True:
|
delay = WORKER_RESPAWN_DELAY
|
||||||
|
while not self._closed:
|
||||||
try:
|
try:
|
||||||
_priority, _seq, future, args = await self._pending.get()
|
await self._add_worker()
|
||||||
except asyncio.CancelledError:
|
except Exception:
|
||||||
|
logger.exception(
|
||||||
|
"Failed to replace preview worker (pool %d/%d); retrying in %ds",
|
||||||
|
len(self._workers),
|
||||||
|
self.size,
|
||||||
|
int(delay),
|
||||||
|
)
|
||||||
|
await asyncio.sleep(delay)
|
||||||
|
delay = min(delay * 2, WORKER_RESPAWN_DELAY_MAX)
|
||||||
|
else:
|
||||||
return
|
return
|
||||||
|
|
||||||
if future.cancelled():
|
async def _dispatch_loop(self) -> None:
|
||||||
continue
|
# Nothing may escape the loop body: a dispatcher that dies silently
|
||||||
|
# permanently shrinks pool capacity and degrades all preview
|
||||||
|
# traffic to timeouts.
|
||||||
|
while True:
|
||||||
try:
|
try:
|
||||||
worker = await asyncio.wait_for(
|
await self._dispatch_one()
|
||||||
self._idle.get(), timeout=PREVIEW_TIMEOUT
|
except asyncio.CancelledError:
|
||||||
)
|
return
|
||||||
except TimeoutError:
|
|
||||||
logger.warning(
|
|
||||||
"Preview worker unavailable (%ds) for %s",
|
|
||||||
int(PREVIEW_TIMEOUT),
|
|
||||||
args[0].name,
|
|
||||||
)
|
|
||||||
if not future.done():
|
|
||||||
future.set_exception(
|
|
||||||
PreviewTimeoutError(
|
|
||||||
args[0].name,
|
|
||||||
backend=_expected_preview_backend(args[0]),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
continue
|
|
||||||
|
|
||||||
filepath = args[0]
|
|
||||||
replace = False
|
|
||||||
try:
|
|
||||||
out, resp = await asyncio.wait_for(
|
|
||||||
worker.request(*args),
|
|
||||||
timeout=PREVIEW_TIMEOUT,
|
|
||||||
)
|
|
||||||
if not future.done():
|
|
||||||
future.set_result((out, resp))
|
|
||||||
except TimeoutError:
|
|
||||||
replace = True
|
|
||||||
if not future.done():
|
|
||||||
future.set_exception(
|
|
||||||
PreviewTimeoutError(
|
|
||||||
filepath.name,
|
|
||||||
backend=_expected_preview_backend(filepath),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
except WorkerChecksumError:
|
|
||||||
replace = True
|
|
||||||
logger.error("Preview checksum mismatch for %s", filepath.name)
|
|
||||||
if not future.done():
|
|
||||||
future.set_exception(
|
|
||||||
PreviewError(f"worker checksum mismatch for {filepath.name}")
|
|
||||||
)
|
|
||||||
except PreviewError as e:
|
|
||||||
if not future.done():
|
|
||||||
future.set_exception(e)
|
|
||||||
except (
|
|
||||||
WorkerProtocolError,
|
|
||||||
asyncio.IncompleteReadError,
|
|
||||||
BrokenPipeError,
|
|
||||||
ConnectionResetError,
|
|
||||||
OSError,
|
|
||||||
ValueError,
|
|
||||||
msgspec.json.DecodeError,
|
|
||||||
) as e:
|
|
||||||
replace = True
|
|
||||||
logger.warning(
|
|
||||||
"Preview worker protocol failure for %s: %s", filepath.name, e
|
|
||||||
)
|
|
||||||
if not future.done():
|
|
||||||
future.set_exception(
|
|
||||||
PreviewError(
|
|
||||||
f"worker protocol failure for {filepath.name}: {e}"
|
|
||||||
)
|
|
||||||
)
|
|
||||||
except Exception:
|
except Exception:
|
||||||
replace = True
|
logger.exception("Preview dispatcher error; continuing")
|
||||||
logger.exception(
|
|
||||||
"Unexpected preview worker error for %s", filepath.name
|
async def _dispatch_one(self) -> None:
|
||||||
)
|
_priority, _seq, future, args = await self._pending.get()
|
||||||
if not future.done():
|
|
||||||
future.set_exception(
|
if future.cancelled():
|
||||||
PreviewError(f"unexpected worker error for {filepath.name}")
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
worker = await asyncio.wait_for(self._idle.get(), timeout=PREVIEW_TIMEOUT)
|
||||||
|
except TimeoutError:
|
||||||
|
logger.warning(
|
||||||
|
"Preview worker unavailable (%ds) for %s",
|
||||||
|
int(PREVIEW_TIMEOUT),
|
||||||
|
args[0].name,
|
||||||
|
)
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(
|
||||||
|
PreviewTimeoutError(
|
||||||
|
args[0].name,
|
||||||
|
backend=_expected_preview_backend(args[0]),
|
||||||
)
|
)
|
||||||
finally:
|
)
|
||||||
if replace:
|
return
|
||||||
await self._replace_worker(worker)
|
|
||||||
elif worker.proc.returncode is None:
|
filepath = args[0]
|
||||||
await self._idle.put(worker)
|
replace = False
|
||||||
else:
|
try:
|
||||||
await self._replace_worker(worker)
|
out, resp = await asyncio.wait_for(
|
||||||
|
worker.request(*args),
|
||||||
|
timeout=PREVIEW_TIMEOUT,
|
||||||
|
)
|
||||||
|
if not future.done():
|
||||||
|
future.set_result((out, resp))
|
||||||
|
except TimeoutError:
|
||||||
|
replace = True
|
||||||
|
logger.warning(
|
||||||
|
"Preview worker pid=%s timed out (%ds) on %s; replacing it",
|
||||||
|
worker.proc.pid,
|
||||||
|
int(PREVIEW_TIMEOUT),
|
||||||
|
filepath.name,
|
||||||
|
)
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(
|
||||||
|
PreviewTimeoutError(
|
||||||
|
filepath.name,
|
||||||
|
backend=_expected_preview_backend(filepath),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
except WorkerChecksumError:
|
||||||
|
replace = True
|
||||||
|
logger.error(
|
||||||
|
"Preview checksum mismatch for %s (worker pid=%s); replacing it",
|
||||||
|
filepath.name,
|
||||||
|
worker.proc.pid,
|
||||||
|
)
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(
|
||||||
|
PreviewError(f"worker checksum mismatch for {filepath.name}")
|
||||||
|
)
|
||||||
|
except PreviewError as e:
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(e)
|
||||||
|
except (
|
||||||
|
WorkerProtocolError,
|
||||||
|
asyncio.IncompleteReadError,
|
||||||
|
BrokenPipeError,
|
||||||
|
ConnectionResetError,
|
||||||
|
OSError,
|
||||||
|
ValueError,
|
||||||
|
msgspec.DecodeError,
|
||||||
|
) as e:
|
||||||
|
replace = True
|
||||||
|
logger.warning(
|
||||||
|
"Preview worker pid=%s protocol failure for %s: %s",
|
||||||
|
worker.proc.pid,
|
||||||
|
filepath.name,
|
||||||
|
e,
|
||||||
|
)
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(
|
||||||
|
PreviewError(f"worker protocol failure for {filepath.name}: {e}")
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
replace = True
|
||||||
|
logger.exception("Unexpected preview worker error for %s", filepath.name)
|
||||||
|
if not future.done():
|
||||||
|
future.set_exception(
|
||||||
|
PreviewError(f"unexpected worker error for {filepath.name}")
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
if replace:
|
||||||
|
await self._replace_worker(worker)
|
||||||
|
elif worker.proc.returncode is None:
|
||||||
|
await self._idle.put(worker)
|
||||||
|
else:
|
||||||
|
await self._replace_worker(worker)
|
||||||
|
|
||||||
async def start(self) -> None:
|
async def start(self) -> None:
|
||||||
workers = await asyncio.gather(
|
workers = await asyncio.gather(
|
||||||
|
|||||||
+65
-28
@@ -17,6 +17,7 @@ import gc
|
|||||||
import io
|
import io
|
||||||
import logging
|
import logging
|
||||||
import mimetypes
|
import mimetypes
|
||||||
|
import os
|
||||||
import shlex
|
import shlex
|
||||||
import struct
|
import struct
|
||||||
import subprocess
|
import subprocess
|
||||||
@@ -26,16 +27,26 @@ from pathlib import Path
|
|||||||
from time import perf_counter
|
from time import perf_counter
|
||||||
|
|
||||||
import av
|
import av
|
||||||
import fitz # PyMuPDF
|
|
||||||
import msgspec
|
import msgspec
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
import pymupdf
|
||||||
import pyvips
|
import pyvips
|
||||||
from blake3 import blake3
|
from blake3 import blake3
|
||||||
|
|
||||||
from cista import config
|
from cista import config
|
||||||
|
from cista.util.logformat import format_level_prefix
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class _WorkerLogFormatter(logging.Formatter):
|
||||||
|
"""Emoji level prefix like the main process, tagged with the worker pid."""
|
||||||
|
|
||||||
|
def format(self, record: logging.LogRecord) -> str:
|
||||||
|
prefix = format_level_prefix(record.levelno)
|
||||||
|
return f"{prefix}worker[{os.getpid()}]: {record.getMessage()}"
|
||||||
|
|
||||||
|
|
||||||
AVIF_FAST_EFFORT = 0
|
AVIF_FAST_EFFORT = 0
|
||||||
|
|
||||||
DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
|
DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
|
||||||
@@ -128,13 +139,20 @@ def _read_request() -> tuple[PreviewRequest, bytes] | None:
|
|||||||
return req, data
|
return req, data
|
||||||
|
|
||||||
|
|
||||||
|
# Raw stdout buffer reserved for the binary protocol once main() redirects
|
||||||
|
# Python-level stdout to stderr. None means "use sys.stdout.buffer as-is"
|
||||||
|
# (CLI single-shot mode, where real stdout is wanted).
|
||||||
|
_protocol_out = None
|
||||||
|
|
||||||
|
|
||||||
def _write_response(resp: PreviewResponse, payload: bytes) -> None:
|
def _write_response(resp: PreviewResponse, payload: bytes) -> None:
|
||||||
|
out = _protocol_out if _protocol_out is not None else sys.stdout.buffer
|
||||||
meta_bytes = _enc.encode(resp)
|
meta_bytes = _enc.encode(resp)
|
||||||
packet = struct.pack("<II", len(meta_bytes), len(payload)) + meta_bytes + payload
|
packet = struct.pack("<II", len(meta_bytes), len(payload)) + meta_bytes + payload
|
||||||
checksum = blake3(packet).digest()
|
checksum = blake3(packet).digest()
|
||||||
sys.stdout.buffer.write(checksum)
|
out.write(checksum)
|
||||||
sys.stdout.buffer.write(packet)
|
out.write(packet)
|
||||||
sys.stdout.buffer.flush()
|
out.flush()
|
||||||
|
|
||||||
|
|
||||||
def dispatch(path, quality, maxsize, maxzoom, data=None):
|
def dispatch(path, quality, maxsize, maxzoom, data=None):
|
||||||
@@ -219,7 +237,18 @@ def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes:
|
|||||||
cmd.insert(5, f"{new_w}x{new_h}")
|
cmd.insert(5, f"{new_w}x{new_h}")
|
||||||
try:
|
try:
|
||||||
try:
|
try:
|
||||||
subprocess.run(cmd, capture_output=True, check=True, shell=False) # noqa: S603
|
# stdin=DEVNULL is critical: ffmpeg must not inherit the worker's
|
||||||
|
# stdin, which carries the framed request protocol. An inherited
|
||||||
|
# stdin lets ffmpeg eat protocol bytes and, if the worker is
|
||||||
|
# killed mid-conversion, keeps the orphaned ffmpeg holding the
|
||||||
|
# pipe open so the parent's proc.wait() hangs forever.
|
||||||
|
subprocess.run( # noqa: S603
|
||||||
|
cmd,
|
||||||
|
capture_output=True,
|
||||||
|
check=True,
|
||||||
|
shell=False,
|
||||||
|
stdin=subprocess.DEVNULL,
|
||||||
|
)
|
||||||
except subprocess.CalledProcessError as e:
|
except subprocess.CalledProcessError as e:
|
||||||
shell_cmd = shlex.join(cmd)
|
shell_cmd = shlex.join(cmd)
|
||||||
stderr = (e.stderr or b"").decode(errors="replace").strip()
|
stderr = (e.stderr or b"").decode(errors="replace").strip()
|
||||||
@@ -270,7 +299,7 @@ def process_image_pyvips(path, *, maxsize, quality):
|
|||||||
".avif",
|
".avif",
|
||||||
Q=quality,
|
Q=quality,
|
||||||
effort=AVIF_FAST_EFFORT,
|
effort=AVIF_FAST_EFFORT,
|
||||||
strip=True,
|
keep="none",
|
||||||
)
|
)
|
||||||
backend = "pyvips"
|
backend = "pyvips"
|
||||||
except pyvips.error.Error:
|
except pyvips.error.Error:
|
||||||
@@ -302,7 +331,7 @@ def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
|
|||||||
".avif",
|
".avif",
|
||||||
Q=quality,
|
Q=quality,
|
||||||
effort=AVIF_FAST_EFFORT,
|
effort=AVIF_FAST_EFFORT,
|
||||||
strip=True,
|
keep="none",
|
||||||
)
|
)
|
||||||
t_end = perf_counter()
|
t_end = perf_counter()
|
||||||
|
|
||||||
@@ -318,19 +347,19 @@ def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
|
|||||||
|
|
||||||
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
||||||
t_load_start = perf_counter()
|
t_load_start = perf_counter()
|
||||||
pdf = fitz.open(path)
|
with pymupdf.open(path) as pdf:
|
||||||
page = pdf.load_page(page_number)
|
page = pdf.load_page(page_number)
|
||||||
w, h = page.rect[2:4]
|
w, h = page.rect[2:4]
|
||||||
zoom = min(maxsize / w, maxsize / h, maxzoom)
|
zoom = min(maxsize / w, maxsize / h, maxzoom)
|
||||||
mat = fitz.Matrix(zoom, zoom)
|
mat = pymupdf.Matrix(zoom, zoom)
|
||||||
pix = page.get_pixmap(matrix=mat)
|
pix = page.get_pixmap(matrix=mat)
|
||||||
t_load_end = perf_counter()
|
t_load_end = perf_counter()
|
||||||
|
|
||||||
t_save_start = perf_counter()
|
t_save_start = perf_counter()
|
||||||
img = pyvips.Image.new_from_memory(
|
img = pyvips.Image.new_from_memory(
|
||||||
pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
|
pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
|
||||||
)
|
)
|
||||||
ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True)
|
ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, keep="none")
|
||||||
backend = "pdf+pyvips"
|
backend = "pdf+pyvips"
|
||||||
t_save_end = perf_counter()
|
t_save_end = perf_counter()
|
||||||
|
|
||||||
@@ -536,24 +565,32 @@ def _run_loop() -> None:
|
|||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
# Configure all log output to stderr before any imports that may emit logs.
|
# Configure all log output to stderr before any imports that may emit
|
||||||
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
|
# logs. stderr is inherited by the parent, so this lands in the server
|
||||||
|
# log, formatted like the main process and tagged with the worker pid.
|
||||||
|
handler = logging.StreamHandler(sys.stderr)
|
||||||
|
handler.setFormatter(_WorkerLogFormatter())
|
||||||
|
logging.basicConfig(level=logging.INFO, handlers=[handler])
|
||||||
|
# pyvips is chatty at INFO ("threadpool completed ..." per operation).
|
||||||
|
logging.getLogger("pyvips").setLevel(logging.WARNING)
|
||||||
try:
|
try:
|
||||||
config.load_config()
|
config.load_config()
|
||||||
logger.warning(
|
|
||||||
"preview-worker config=%s master_secret=%s",
|
|
||||||
config.conffile,
|
|
||||||
config.config.secret,
|
|
||||||
)
|
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("preview-worker failed to load config at startup")
|
logger.exception("preview-worker failed to load config at startup")
|
||||||
if len(sys.argv) > 1:
|
if len(sys.argv) > 1:
|
||||||
_run_once()
|
_run_once()
|
||||||
return
|
return
|
||||||
|
# The command channel is a binary protocol on fd 1. Anything printed to
|
||||||
|
# stdout by Python code (e.g. a library emitting a warning via print())
|
||||||
|
# would corrupt the protocol, so redirect Python-level stdout to stderr
|
||||||
|
# (the server log) and keep the raw buffer solely for protocol traffic.
|
||||||
|
global _protocol_out
|
||||||
|
_protocol_out = sys.stdout.buffer
|
||||||
|
sys.stdout = sys.stderr
|
||||||
# Eagerly import heavy modules before signalling readiness so the parent
|
# Eagerly import heavy modules before signalling readiness so the parent
|
||||||
# does not hand us a request while we are still initialising.
|
# does not hand us a request while we are still initialising.
|
||||||
sys.stdout.buffer.write(b"\x01")
|
_protocol_out.write(b"\x01")
|
||||||
sys.stdout.buffer.flush()
|
_protocol_out.flush()
|
||||||
_run_loop()
|
_run_loop()
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+4
-31
@@ -3,11 +3,13 @@
|
|||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
import unicodedata
|
|
||||||
from ipaddress import IPv6Address
|
from ipaddress import IPv6Address
|
||||||
|
|
||||||
from sanic.log import LOGGING_CONFIG_DEFAULTS
|
from sanic.log import LOGGING_CONFIG_DEFAULTS
|
||||||
|
|
||||||
|
from cista.util.logformat import EmojiFormatter as _EmojiFormatter
|
||||||
|
from cista.util.logformat import display_width as _display_width
|
||||||
|
|
||||||
logger = logging.getLogger("cista.access")
|
logger = logging.getLogger("cista.access")
|
||||||
|
|
||||||
|
|
||||||
@@ -132,14 +134,6 @@ def format_duration_ms(duration_ms: float) -> str:
|
|||||||
return f"{hours}h{minutes}m"
|
return f"{hours}h{minutes}m"
|
||||||
|
|
||||||
|
|
||||||
def _display_width(text: str) -> int:
|
|
||||||
return sum(
|
|
||||||
1 + (unicodedata.east_asian_width(c) in "FW")
|
|
||||||
for c in text
|
|
||||||
if unicodedata.category(c) != "Mn"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _format_left(label: str) -> str:
|
def _format_left(label: str) -> str:
|
||||||
return label[:19].ljust(19)
|
return label[:19].ljust(19)
|
||||||
|
|
||||||
@@ -156,6 +150,7 @@ def format_access_log(
|
|||||||
method: str,
|
method: str,
|
||||||
host: str,
|
host: str,
|
||||||
path: str,
|
path: str,
|
||||||
|
*,
|
||||||
duration_ms: float,
|
duration_ms: float,
|
||||||
extra: str | None = None,
|
extra: str | None = None,
|
||||||
) -> str:
|
) -> str:
|
||||||
@@ -278,28 +273,6 @@ def configure_access_logging() -> None:
|
|||||||
logger.propagate = False
|
logger.propagate = False
|
||||||
|
|
||||||
|
|
||||||
_LEVEL_EMOJI = {
|
|
||||||
logging.DEBUG: "🔍",
|
|
||||||
logging.INFO: "ℹ️", # noqa: RUF001
|
|
||||||
logging.WARNING: "⚠️",
|
|
||||||
logging.ERROR: "🛑",
|
|
||||||
logging.CRITICAL: "🛑",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _format_level_prefix(levelno: int) -> str:
|
|
||||||
emoji = _LEVEL_EMOJI.get(levelno, "▪️")
|
|
||||||
prefix = f"{emoji} "
|
|
||||||
return prefix + (" " * max(0, 3 - _display_width(prefix)))
|
|
||||||
|
|
||||||
|
|
||||||
class _EmojiFormatter(logging.Formatter):
|
|
||||||
"""Compact formatter: emoji + message, no timestamp/level text/logger name."""
|
|
||||||
|
|
||||||
def format(self, record: logging.LogRecord) -> str:
|
|
||||||
return _format_level_prefix(record.levelno) + record.getMessage()
|
|
||||||
|
|
||||||
|
|
||||||
def configure_main_logging() -> None:
|
def configure_main_logging() -> None:
|
||||||
"""Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format.
|
"""Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format.
|
||||||
|
|
||||||
|
|||||||
+11
-1
@@ -62,12 +62,18 @@ async def close_client():
|
|||||||
_client = None
|
_client = None
|
||||||
|
|
||||||
|
|
||||||
async def validate_sso_request(request, *, perm: str = "cista:login") -> dict | None:
|
async def validate_sso_request(
|
||||||
|
request, *, perm: str = "cista:login", renew: bool = True
|
||||||
|
) -> dict | None:
|
||||||
"""Validate an SSO request against the auth backend.
|
"""Validate an SSO request against the auth backend.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
request: The Sanic request object
|
request: The Sanic request object
|
||||||
perm: Permission to validate (default: cista:login, privileged also cista:admin)
|
perm: Permission to validate (default: cista:login, privileged also cista:admin)
|
||||||
|
renew: Whether to allow the auth backend to renew the session cookie.
|
||||||
|
Use ``False`` for WebSocket validation where Set-Cookie cannot be
|
||||||
|
forwarded to the client; this makes the request read-only and avoids
|
||||||
|
resetting the backend renewal timeout.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
User info dict if valid, None if validation fails with auth required response
|
User info dict if valid, None if validation fails with auth required response
|
||||||
@@ -88,12 +94,16 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
|
|||||||
headers["cookie"] = request.headers["cookie"]
|
headers["cookie"] = request.headers["cookie"]
|
||||||
if "authorization" in request.headers:
|
if "authorization" in request.headers:
|
||||||
headers["authorization"] = request.headers["authorization"]
|
headers["authorization"] = request.headers["authorization"]
|
||||||
|
if "user-agent" in request.headers:
|
||||||
|
headers["user-agent"] = request.headers["user-agent"]
|
||||||
headers["accept"] = "application/json"
|
headers["accept"] = "application/json"
|
||||||
headers["x-forwarded-for"] = request.client_ip
|
headers["x-forwarded-for"] = request.client_ip
|
||||||
headers["x-forwarded-host"] = request.host
|
headers["x-forwarded-host"] = request.host
|
||||||
headers["x-forwarded-proto"] = request.scheme
|
headers["x-forwarded-proto"] = request.scheme
|
||||||
|
|
||||||
url = f"{PASKIA_BACKEND_URL}/auth/api/validate?perm={perm}"
|
url = f"{PASKIA_BACKEND_URL}/auth/api/validate?perm={perm}"
|
||||||
|
if not renew:
|
||||||
|
url += "&renew=0"
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response = await client.post(
|
response = await client.post(
|
||||||
|
|||||||
@@ -0,0 +1,37 @@
|
|||||||
|
"""Shared log formatting helpers with no Sanic dependency.
|
||||||
|
|
||||||
|
Used by the main process (cista.sanic_logging) and by the preview worker
|
||||||
|
subprocess, which must not import Sanic.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import unicodedata
|
||||||
|
|
||||||
|
LEVEL_EMOJI = {
|
||||||
|
logging.DEBUG: "🔍",
|
||||||
|
logging.INFO: "ℹ️", # noqa: RUF001
|
||||||
|
logging.WARNING: "⚠️",
|
||||||
|
logging.ERROR: "🛑",
|
||||||
|
logging.CRITICAL: "🛑",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def display_width(text: str) -> int:
|
||||||
|
return sum(
|
||||||
|
1 + (unicodedata.east_asian_width(c) in "FW")
|
||||||
|
for c in text
|
||||||
|
if unicodedata.category(c) != "Mn"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def format_level_prefix(levelno: int) -> str:
|
||||||
|
emoji = LEVEL_EMOJI.get(levelno, "▪️")
|
||||||
|
prefix = f"{emoji} "
|
||||||
|
return prefix + (" " * max(0, 3 - display_width(prefix)))
|
||||||
|
|
||||||
|
|
||||||
|
class EmojiFormatter(logging.Formatter):
|
||||||
|
"""Compact formatter: emoji + message, no timestamp/level text/logger name."""
|
||||||
|
|
||||||
|
def format(self, record: logging.LogRecord) -> str:
|
||||||
|
return format_level_prefix(record.levelno) + record.getMessage()
|
||||||
@@ -6,9 +6,8 @@
|
|||||||
"dev": "vite",
|
"dev": "vite",
|
||||||
"build": "run-p type-check \"build-only {@}\" --",
|
"build": "run-p type-check \"build-only {@}\" --",
|
||||||
"preview": "vite preview",
|
"preview": "vite preview",
|
||||||
"test:unit": "vitest",
|
|
||||||
"build-only": "vite build",
|
"build-only": "vite build",
|
||||||
"type-check": "vue-tsc --noEmit -p tsconfig.vitest.json --composite false",
|
"type-check": "vue-tsc --noEmit -p tsconfig.app.json --composite false",
|
||||||
"lint": "biome lint .",
|
"lint": "biome lint .",
|
||||||
"format": "biome format --write .",
|
"format": "biome format --write .",
|
||||||
"format:check": "biome format --check .",
|
"format:check": "biome format --check .",
|
||||||
@@ -37,17 +36,13 @@
|
|||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
"@biomejs/biome": "^1.9.4",
|
"@biomejs/biome": "^1.9.4",
|
||||||
"@tsconfig/node18": "^18.2.6",
|
"@tsconfig/node18": "^18.2.6",
|
||||||
"@types/jsdom": "^27.0.0",
|
|
||||||
"@types/lodash-es": "^4.17.12",
|
"@types/lodash-es": "^4.17.12",
|
||||||
"@types/node": "^25.1.0",
|
"@types/node": "^25.1.0",
|
||||||
"@vitejs/plugin-vue": "^6.0.3",
|
"@vitejs/plugin-vue": "^6.0.3",
|
||||||
"@vue/test-utils": "^2.4.6",
|
|
||||||
"@vue/tsconfig": "^0.8.1",
|
"@vue/tsconfig": "^0.8.1",
|
||||||
"jsdom": "^27.4.0",
|
|
||||||
"npm-run-all2": "^8.0.4",
|
"npm-run-all2": "^8.0.4",
|
||||||
"typescript": "~5.9.3",
|
"typescript": "~5.9.3",
|
||||||
"vite": "^7.3.1",
|
"vite": "^7.3.1",
|
||||||
"vitest": "^4.0.18",
|
|
||||||
"vue-tsc": "^3.2.4"
|
"vue-tsc": "^3.2.4"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -53,7 +53,7 @@
|
|||||||
<a :href="doc.text ? doc.editurl : doc.url" tabindex=-1 @contextmenu.stop @focus.stop="store.cursor = doc.key">
|
<a :href="doc.text ? doc.editurl : doc.url" tabindex=-1 @contextmenu.stop @focus.stop="store.cursor = doc.key">
|
||||||
{{ doc.name }}
|
{{ doc.name }}
|
||||||
</a>
|
</a>
|
||||||
<button tabindex=-1 v-if="store.cursor == doc.key" class="rename-button" @click="() => (editing = doc)">🖊️</button>
|
<button tabindex=-1 class="rename-button" @click="() => (editing = doc)">🖊️</button>
|
||||||
</template>
|
</template>
|
||||||
</td>
|
</td>
|
||||||
<FileModified :doc=doc :now=nowkey />
|
<FileModified :doc=doc :now=nowkey />
|
||||||
@@ -84,6 +84,7 @@ import ContextMenu from '@imengyu/vue3-context-menu'
|
|||||||
import {
|
import {
|
||||||
computed,
|
computed,
|
||||||
nextTick,
|
nextTick,
|
||||||
|
onActivated,
|
||||||
onDeactivated,
|
onDeactivated,
|
||||||
onMounted,
|
onMounted,
|
||||||
onUnmounted,
|
onUnmounted,
|
||||||
@@ -336,9 +337,13 @@ const focusBreadcrumb = () => {
|
|||||||
const keyboardFollowScroll = createKeyboardFollowScroll()
|
const keyboardFollowScroll = createKeyboardFollowScroll()
|
||||||
const markKeyboardFollow = keyboardFollowScroll.markKeyboardFollow
|
const markKeyboardFollow = keyboardFollowScroll.markKeyboardFollow
|
||||||
const keepCursorVisibleSmooth = keyboardFollowScroll.keepVisible
|
const keepCursorVisibleSmooth = keyboardFollowScroll.keepVisible
|
||||||
|
// Deactivated (KeepAlive-cached) instances stay alive with frozen, potentially
|
||||||
|
// stale props - their watchers must not react to global store changes.
|
||||||
|
let isActive = true
|
||||||
watch(
|
watch(
|
||||||
() => store.cursor,
|
() => store.cursor,
|
||||||
cursor => {
|
cursor => {
|
||||||
|
if (!isActive) return
|
||||||
if (cursor && editing.value && cursor !== editing.value.key) {
|
if (cursor && editing.value && cursor !== editing.value.key) {
|
||||||
exitEditing()
|
exitEditing()
|
||||||
}
|
}
|
||||||
@@ -347,6 +352,7 @@ watch(
|
|||||||
watch(
|
watch(
|
||||||
() => store.cursor,
|
() => store.cursor,
|
||||||
cursor => {
|
cursor => {
|
||||||
|
if (!isActive) return
|
||||||
if (cursor && !editing.value) {
|
if (cursor && !editing.value) {
|
||||||
const a = document.querySelector(
|
const a = document.querySelector(
|
||||||
`#file-${cursor} .name a`
|
`#file-${cursor} .name a`
|
||||||
@@ -359,6 +365,7 @@ watch(
|
|||||||
watch(
|
watch(
|
||||||
() => [props.documents.length, store.cursor, store.query, editing.value] as const,
|
() => [props.documents.length, store.cursor, store.query, editing.value] as const,
|
||||||
([len, cursor, query, editingDoc]) => {
|
([len, cursor, query, editingDoc]) => {
|
||||||
|
if (!isActive) return
|
||||||
if (!len && cursor && !query && !editingDoc) {
|
if (!len && cursor && !query && !editingDoc) {
|
||||||
store.cursor = ''
|
store.cursor = ''
|
||||||
focusBreadcrumb()
|
focusBreadcrumb()
|
||||||
@@ -378,7 +385,11 @@ onMounted(() => {
|
|||||||
active.focus({ preventScroll: true })
|
active.focus({ preventScroll: true })
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
onActivated(() => {
|
||||||
|
isActive = true
|
||||||
|
})
|
||||||
onDeactivated(() => {
|
onDeactivated(() => {
|
||||||
|
isActive = false
|
||||||
if (editing.value) exitEditing()
|
if (editing.value) exitEditing()
|
||||||
})
|
})
|
||||||
onUnmounted(() => {
|
onUnmounted(() => {
|
||||||
@@ -617,6 +628,12 @@ table td {
|
|||||||
.name .rename-button {
|
.name .rename-button {
|
||||||
position: absolute;
|
position: absolute;
|
||||||
right: 0;
|
right: 0;
|
||||||
|
opacity: 0;
|
||||||
|
visibility: hidden;
|
||||||
|
}
|
||||||
|
tbody tr:hover .name .rename-button {
|
||||||
|
opacity: 1;
|
||||||
|
visibility: visible;
|
||||||
animation: appear calc(5 * var(--transition-time)) linear;
|
animation: appear calc(5 * var(--transition-time)) linear;
|
||||||
}
|
}
|
||||||
@keyframes appear {
|
@keyframes appear {
|
||||||
|
|||||||
@@ -8,7 +8,7 @@
|
|||||||
:editing="editing === doc ? {rename, exit} : null"
|
:editing="editing === doc ? {rename, exit} : null"
|
||||||
:style="{ '--gallery-figure-height': rowHeightsByKey[doc.key] ?? '15em' }"
|
:style="{ '--gallery-figure-height': rowHeightsByKey[doc.key] ?? '15em' }"
|
||||||
@menu="contextMenu($event, doc)"
|
@menu="contextMenu($event, doc)"
|
||||||
@rename="editing = doc; store.cursor = doc.key"
|
@rename="onFigureRename(doc)"
|
||||||
:class="{ 'folder-start': showFolderBreadcrumb(index) }"
|
:class="{ 'folder-start': showFolderBreadcrumb(index) }"
|
||||||
/>
|
/>
|
||||||
</template>
|
</template>
|
||||||
@@ -64,6 +64,10 @@ const editing = shallowRef<Doc | null>(null)
|
|||||||
const exit = () => {
|
const exit = () => {
|
||||||
editing.value = null
|
editing.value = null
|
||||||
}
|
}
|
||||||
|
const onFigureRename = (doc: Doc) => {
|
||||||
|
editing.value = doc
|
||||||
|
store.cursor = doc.key
|
||||||
|
}
|
||||||
const rename = async (doc: Doc, newName: string) => {
|
const rename = async (doc: Doc, newName: string) => {
|
||||||
const oldName = doc.name
|
const oldName = doc.name
|
||||||
doc.name = newName // We should get an update from watch but this is quicker
|
doc.name = newName // We should get an update from watch but this is quicker
|
||||||
@@ -396,9 +400,13 @@ const focusBreadcrumb = () => {
|
|||||||
const keyboardFollowScroll = createKeyboardFollowScroll()
|
const keyboardFollowScroll = createKeyboardFollowScroll()
|
||||||
const markKeyboardFollow = keyboardFollowScroll.markKeyboardFollow
|
const markKeyboardFollow = keyboardFollowScroll.markKeyboardFollow
|
||||||
const keepCursorVisibleSmooth = keyboardFollowScroll.keepVisible
|
const keepCursorVisibleSmooth = keyboardFollowScroll.keepVisible
|
||||||
|
// Deactivated (KeepAlive-cached) instances stay alive with frozen, potentially
|
||||||
|
// stale props - their watchers must not react to global store changes.
|
||||||
|
let isActive = true
|
||||||
watch(
|
watch(
|
||||||
() => store.cursor,
|
() => store.cursor,
|
||||||
cursor => {
|
cursor => {
|
||||||
|
if (!isActive) return
|
||||||
if (cursor && editing.value && cursor !== editing.value.key) {
|
if (cursor && editing.value && cursor !== editing.value.key) {
|
||||||
exit()
|
exit()
|
||||||
}
|
}
|
||||||
@@ -407,6 +415,7 @@ watch(
|
|||||||
watch(
|
watch(
|
||||||
() => store.cursor,
|
() => store.cursor,
|
||||||
cursor => {
|
cursor => {
|
||||||
|
if (!isActive) return
|
||||||
if (cursor && !editing.value) {
|
if (cursor && !editing.value) {
|
||||||
const a = document.querySelector(`#file-${cursor}`) as HTMLAnchorElement | null
|
const a = document.querySelector(`#file-${cursor}`) as HTMLAnchorElement | null
|
||||||
if (a) {
|
if (a) {
|
||||||
@@ -419,6 +428,7 @@ watch(
|
|||||||
watch(
|
watch(
|
||||||
() => [props.documents.length, store.cursor, store.query, editing.value] as const,
|
() => [props.documents.length, store.cursor, store.query, editing.value] as const,
|
||||||
([len, cursor, query, editingDoc]) => {
|
([len, cursor, query, editingDoc]) => {
|
||||||
|
if (!isActive) return
|
||||||
if (!len && cursor && !query && !editingDoc) {
|
if (!len && cursor && !query && !editingDoc) {
|
||||||
store.cursor = ''
|
store.cursor = ''
|
||||||
focusBreadcrumb()
|
focusBreadcrumb()
|
||||||
@@ -449,12 +459,14 @@ onMounted(() => {
|
|||||||
attachGalleryObservers()
|
attachGalleryObservers()
|
||||||
})
|
})
|
||||||
onActivated(() => {
|
onActivated(() => {
|
||||||
|
isActive = true
|
||||||
nextTick(() => {
|
nextTick(() => {
|
||||||
updateColumns()
|
updateColumns()
|
||||||
attachGalleryObservers()
|
attachGalleryObservers()
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
onDeactivated(() => {
|
onDeactivated(() => {
|
||||||
|
isActive = false
|
||||||
detachGalleryObservers()
|
detachGalleryObservers()
|
||||||
if (editing.value) exit()
|
if (editing.value) exit()
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -29,7 +29,7 @@
|
|||||||
<span class="filename">{{ snap.displayName }}<SparseIndicator :doc="doc" class="after-name" /></span>
|
<span class="filename">{{ snap.displayName }}<SparseIndicator :doc="doc" class="after-name" /></span>
|
||||||
<span v-if="snap.ext" class="file-ext">.{{ snap.ext }}</span>
|
<span v-if="snap.ext" class="file-ext">.{{ snap.ext }}</span>
|
||||||
</span>
|
</span>
|
||||||
<button class="rename-btn" @click="$emit('rename')" title="Rename">✏️</button>
|
<button class="rename-btn" @click="emit('rename')" title="Rename">✏️</button>
|
||||||
</div>
|
</div>
|
||||||
<div class=namespacer></div>
|
<div class=namespacer></div>
|
||||||
</template>
|
</template>
|
||||||
@@ -64,6 +64,10 @@ const props = defineProps<{
|
|||||||
doc: Doc
|
doc: Doc
|
||||||
editing?: EditingProp
|
editing?: EditingProp
|
||||||
}>()
|
}>()
|
||||||
|
const emit = defineEmits<{
|
||||||
|
(e: 'rename'): void
|
||||||
|
(e: 'menu', ev: MouseEvent): void
|
||||||
|
}>()
|
||||||
const m = ref<typeof MediaPreview | null>(null)
|
const m = ref<typeof MediaPreview | null>(null)
|
||||||
const tooltip = ref<InstanceType<typeof CursorTooltip> | null>(null)
|
const tooltip = ref<InstanceType<typeof CursorTooltip> | null>(null)
|
||||||
|
|
||||||
|
|||||||
@@ -6,9 +6,6 @@
|
|||||||
},
|
},
|
||||||
{
|
{
|
||||||
"path": "./tsconfig.app.json"
|
"path": "./tsconfig.app.json"
|
||||||
},
|
|
||||||
{
|
|
||||||
"path": "./tsconfig.vitest.json"
|
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,12 +1,6 @@
|
|||||||
{
|
{
|
||||||
"extends": "@tsconfig/node18/tsconfig.json",
|
"extends": "@tsconfig/node18/tsconfig.json",
|
||||||
"include": [
|
"include": ["vite.config.*"],
|
||||||
"vite.config.*",
|
|
||||||
"vitest.config.*",
|
|
||||||
"cypress.config.*",
|
|
||||||
"nightwatch.conf.*",
|
|
||||||
"playwright.config.*"
|
|
||||||
],
|
|
||||||
"compilerOptions": {
|
"compilerOptions": {
|
||||||
"composite": true,
|
"composite": true,
|
||||||
"module": "ESNext",
|
"module": "ESNext",
|
||||||
|
|||||||
@@ -1,8 +0,0 @@
|
|||||||
{
|
|
||||||
"extends": "./tsconfig.app.json",
|
|
||||||
"exclude": [],
|
|
||||||
"compilerOptions": {
|
|
||||||
"composite": true,
|
|
||||||
"types": ["node", "jsdom"]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+2
-1
@@ -132,6 +132,7 @@ ignore = [
|
|||||||
"ANN205", # legacy codebase: no full runtime annotation coverage yet
|
"ANN205", # legacy codebase: no full runtime annotation coverage yet
|
||||||
"BLE001", # broad catch remains in boundary/proxy/error-handling paths
|
"BLE001", # broad catch remains in boundary/proxy/error-handling paths
|
||||||
"C901", # legacy complexity; keep other correctness rules enabled
|
"C901", # legacy complexity; keep other correctness rules enabled
|
||||||
|
"CPY", # copyright notices not wanted in this codebase
|
||||||
"D100", # legacy docs not yet standardized
|
"D100", # legacy docs not yet standardized
|
||||||
"D101", # legacy docs not yet standardized
|
"D101", # legacy docs not yet standardized
|
||||||
"D102", # legacy docs not yet standardized
|
"D102", # legacy docs not yet standardized
|
||||||
@@ -160,7 +161,7 @@ ignore = [
|
|||||||
"TRY003", # exception-message strictness too noisy on legacy handlers
|
"TRY003", # exception-message strictness too noisy on legacy handlers
|
||||||
]
|
]
|
||||||
isort.known-first-party = ["cista"]
|
isort.known-first-party = ["cista"]
|
||||||
per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001"]
|
per-file-ignores."tests/*" = ["S", "ANN", "D", "INP", "PLR2004", "ARG001", "SLF001"]
|
||||||
per-file-ignores."scripts/*" = ["T20"]
|
per-file-ignores."scripts/*" = ["T20"]
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
|
|||||||
@@ -0,0 +1,175 @@
|
|||||||
|
"""Tests for the preview worker pool resilience.
|
||||||
|
|
||||||
|
Regression context: a piped worker stderr that nobody drains used to block
|
||||||
|
the worker mid-request once the OS pipe buffer filled, and asyncio's
|
||||||
|
proc.wait() then never resolved even after kill() — wedging one dispatcher
|
||||||
|
per stuck worker until all preview traffic timed out permanently.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import sys
|
||||||
|
import textwrap
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import AsyncMock, Mock
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from cista import preview
|
||||||
|
|
||||||
|
FAKE_WORKER = textwrap.dedent(
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import struct
|
||||||
|
import sys
|
||||||
|
|
||||||
|
import blake3
|
||||||
|
|
||||||
|
|
||||||
|
def read_exact(n):
|
||||||
|
buf = b""
|
||||||
|
while len(buf) < n:
|
||||||
|
chunk = sys.stdin.buffer.read(n - len(buf))
|
||||||
|
if not chunk:
|
||||||
|
raise EOFError
|
||||||
|
buf += chunk
|
||||||
|
return buf
|
||||||
|
|
||||||
|
|
||||||
|
sys.stdout.buffer.write(b"\\x01")
|
||||||
|
sys.stdout.buffer.flush()
|
||||||
|
while True:
|
||||||
|
header = sys.stdin.buffer.read(8)
|
||||||
|
if not header or len(header) < 8:
|
||||||
|
break
|
||||||
|
meta_len, payload_len = struct.unpack("<II", header)
|
||||||
|
meta = read_exact(meta_len)
|
||||||
|
read_exact(payload_len)
|
||||||
|
req = json.loads(meta)
|
||||||
|
if req["path"].endswith(".block"):
|
||||||
|
# Simulate a worker stuck on an undrained stderr pipe:
|
||||||
|
# flood stderr past the OS pipe buffer, then never respond.
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
|
||||||
|
try:
|
||||||
|
os.write(2, b"x" * 10_000_000)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
while True:
|
||||||
|
time.sleep(3600)
|
||||||
|
resp = json.dumps({"ok": True, "mime": "image/avif", "backend": "fake"}).encode()
|
||||||
|
payload = b"FAKEIMG"
|
||||||
|
packet = struct.pack("<II", len(resp), len(payload)) + resp + payload
|
||||||
|
sys.stdout.buffer.write(blake3.blake3(packet).digest())
|
||||||
|
sys.stdout.buffer.write(packet)
|
||||||
|
sys.stdout.buffer.flush()
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_pool_recovers_from_wedged_worker(monkeypatch, tmp_path):
|
||||||
|
"""A worker wedged on an undrained stderr pipe must not kill the pool.
|
||||||
|
|
||||||
|
Recreates the old production setup (stderr=PIPE, never drained) and
|
||||||
|
verifies the request times out, the stuck worker's kill() cannot hang
|
||||||
|
the dispatcher, and the pool serves the next request normally.
|
||||||
|
"""
|
||||||
|
monkeypatch.setattr(preview, "PREVIEW_TIMEOUT", 1.0)
|
||||||
|
monkeypatch.setattr(preview, "WORKER_KILL_GRACE", 0.5)
|
||||||
|
monkeypatch.setattr(preview, "WORKER_RESPAWN_DELAY", 0.05)
|
||||||
|
script = tmp_path / "fake_worker.py"
|
||||||
|
script.write_text(FAKE_WORKER)
|
||||||
|
|
||||||
|
async def fake_spawn(self):
|
||||||
|
proc = await asyncio.create_subprocess_exec(
|
||||||
|
sys.executable,
|
||||||
|
str(script),
|
||||||
|
stdin=asyncio.subprocess.PIPE,
|
||||||
|
stdout=asyncio.subprocess.PIPE,
|
||||||
|
# Deliberately piped-and-undrained, recreating the old
|
||||||
|
# production setup that wedges a worker on stderr writes.
|
||||||
|
stderr=asyncio.subprocess.PIPE,
|
||||||
|
)
|
||||||
|
preview._active_procs.add(proc)
|
||||||
|
await asyncio.wait_for(proc.stdout.readexactly(1), timeout=10)
|
||||||
|
return preview._PreviewWorker(proc)
|
||||||
|
|
||||||
|
monkeypatch.setattr(preview._PreviewWorkerPool, "_spawn_worker", fake_spawn)
|
||||||
|
|
||||||
|
pool = preview._PreviewWorkerPool(1)
|
||||||
|
await pool.start()
|
||||||
|
try:
|
||||||
|
with pytest.raises(preview.PreviewTimeoutError):
|
||||||
|
await pool.run(Path("wedged.block"), 60, 512, 2.0)
|
||||||
|
|
||||||
|
out, resp = await asyncio.wait_for(
|
||||||
|
pool.run(Path("ok.jpg"), 60, 512, 2.0), timeout=10
|
||||||
|
)
|
||||||
|
assert out == b"FAKEIMG"
|
||||||
|
assert resp.ok
|
||||||
|
assert all(not task.done() for task in pool._dispatchers)
|
||||||
|
finally:
|
||||||
|
await pool.close()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_worker_kill_grace_when_wait_hangs(monkeypatch):
|
||||||
|
"""kill() must return even if asyncio never resolves proc.wait()."""
|
||||||
|
monkeypatch.setattr(preview, "WORKER_KILL_GRACE", 0.1)
|
||||||
|
proc = Mock()
|
||||||
|
proc.returncode = None
|
||||||
|
proc.pid = 1234
|
||||||
|
never = asyncio.Future()
|
||||||
|
|
||||||
|
async def wait():
|
||||||
|
await never
|
||||||
|
|
||||||
|
proc.wait = wait
|
||||||
|
worker = preview._PreviewWorker(proc)
|
||||||
|
preview._active_procs.add(proc)
|
||||||
|
start = time.monotonic()
|
||||||
|
await worker.kill()
|
||||||
|
assert time.monotonic() - start < 2
|
||||||
|
assert proc not in preview._active_procs
|
||||||
|
never.cancel()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_replace_worker_retries_failed_spawn(monkeypatch):
|
||||||
|
"""A failed replacement spawn must be retried, not silently dropped."""
|
||||||
|
monkeypatch.setattr(preview, "WORKER_RESPAWN_DELAY", 0.01)
|
||||||
|
pool = preview._PreviewWorkerPool(1)
|
||||||
|
old_worker = Mock()
|
||||||
|
old_worker.proc = Mock(pid=4321)
|
||||||
|
old_worker.kill = AsyncMock()
|
||||||
|
attempts = 0
|
||||||
|
|
||||||
|
async def add_worker():
|
||||||
|
nonlocal attempts
|
||||||
|
attempts += 1
|
||||||
|
if attempts < 3:
|
||||||
|
raise OSError("too many open files")
|
||||||
|
|
||||||
|
pool._add_worker = add_worker
|
||||||
|
await pool._replace_worker(old_worker)
|
||||||
|
assert attempts == 3
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_loop_survives_body_errors():
|
||||||
|
"""Exceptions escaping a dispatch cycle must not kill the dispatcher."""
|
||||||
|
pool = preview._PreviewWorkerPool(1)
|
||||||
|
calls = 0
|
||||||
|
|
||||||
|
async def dispatch_one():
|
||||||
|
nonlocal calls
|
||||||
|
calls += 1
|
||||||
|
if calls == 1:
|
||||||
|
raise RuntimeError("boom")
|
||||||
|
raise asyncio.CancelledError
|
||||||
|
|
||||||
|
pool._dispatch_one = dispatch_one
|
||||||
|
await pool._dispatch_loop()
|
||||||
|
assert calls == 2
|
||||||
Reference in New Issue
Block a user