pristontale-eu-linux/ptlinux/locking.py

"""Cross-process locks for launcher-prefix mutations."""

from __future__ import annotations

from contextlib import contextmanager
from datetime import datetime
import fcntl
import os
from pathlib import Path
from typing import Callable, Iterator, Sequence, TextIO


ActiveGameProbe = Callable[[], bool]
MAX_PROCESS_ENVIRONMENT_BYTES = 1024 * 1024
MAX_PROCESS_COMMAND_BYTES = 128 * 1024


class OperationBusyError(RuntimeError):
	"""Raised when another launcher-prefix operation is still active."""


class LauncherPlayLease:
	"""Exclusive preparation lease that can become a shared game lease."""

	def __init__(
		self,
		handle: TextIO | None,
		*,
		preparation_allowed: bool,
		record_owned: bool,
	) -> None:
		self._handle = handle
		self.preparation_allowed = preparation_allowed
		self._record_owned = record_owned
		self._shared = not preparation_allowed

	@classmethod
	def dry_run(cls) -> LauncherPlayLease:
		return cls(None, preparation_allowed=True, record_owned=False)

	@property
	def shared(self) -> bool:
		return self._shared

	def allow_shared_clients(self) -> None:
		"""Downgrade after all runtime mutations have completed."""

		if self._shared:
			return
		if self._handle is not None:
			_write_owner(self._handle, "play")
			fcntl.flock(self._handle.fileno(), fcntl.LOCK_SH)
		self._shared = True

	@property
	def record_owned(self) -> bool:
		return self._record_owned

	@property
	def inherited_fds(self) -> tuple[int, ...]:
		"""Keep the shared lease alive in the launched game supervisor."""

		if self._handle is None:
			return ()
		return (self._handle.fileno(),)


@contextmanager
def launcher_operation_lock(
	lock_path: Path,
	operation: str,
	*,
	active_game_probe: ActiveGameProbe | None = None,
) -> Iterator[None]:
	lock_path.parent.mkdir(parents=True, exist_ok=True)
	with lock_path.open("a+", encoding="utf-8") as handle:
		try:
			fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
		except BlockingIOError as exc:
			handle.seek(0)
			owner = handle.read().strip() or "unknown operation"
			raise OperationBusyError(
				f"launcher files are busy ({owner}); wait for it to finish and try again"
			) from exc
		if active_game_probe is not None and active_game_probe():
			raise OperationBusyError(
				"launcher files are busy (active managed game client detected); "
				"close every game client and try again"
			)
		_write_owner(handle, operation)
		try:
			yield
		finally:
			handle.seek(0)
			handle.truncate()
			handle.flush()
			fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


@contextmanager
def launcher_play_lock(
	lock_path: Path,
	operation: str = "play",
	*,
	active_game_probe: ActiveGameProbe | None = None,
) -> Iterator[LauncherPlayLease]:
	"""Serialize preparation, then allow multiple active game clients.

	The first play process acquires an exclusive lease so it may safely rebuild
	the runtime view or prefix. It downgrades that lease immediately before
	starting Game.exe. Later play processes then receive shared, read-only leases,
	while updates and repairs remain excluded until every client exits.
	"""

	lock_path.parent.mkdir(parents=True, exist_ok=True)
	with lock_path.open("a+", encoding="utf-8") as handle:
		try:
			fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
		except BlockingIOError:
			try:
				fcntl.flock(handle.fileno(), fcntl.LOCK_SH | fcntl.LOCK_NB)
			except BlockingIOError as exc:
				owner = _read_owner(handle)
				raise OperationBusyError(
					f"launcher files are busy ({owner}); wait for it to finish and try again"
				) from exc
			lease = LauncherPlayLease(
				handle,
				preparation_allowed=False,
				record_owned=False,
			)
		else:
			if active_game_probe is not None and active_game_probe():
				# A game from an older launcher may have outlived its Python
				# supervisor and therefore its inherited lock. Rejoin that live
				# prefix without mutating its runtime view or waiting for it to exit.
				_write_owner(handle, operation)
				fcntl.flock(handle.fileno(), fcntl.LOCK_SH)
				lease = LauncherPlayLease(
					handle,
					preparation_allowed=False,
					record_owned=True,
				)
			else:
				_write_owner(handle, f"{operation} preparation")
				lease = LauncherPlayLease(
					handle,
					preparation_allowed=True,
					record_owned=True,
				)
		try:
			yield lease
		finally:
			# A downgraded owner record remains useful while any secondary shared
			# lease is active. The next exclusive operation atomically replaces it.
			if lease.record_owned and not lease.shared:
				handle.seek(0)
				handle.truncate()
				handle.flush()
			fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


def managed_game_process_is_active(
	runtime_view: Path,
	wine_prefixes: Sequence[Path],
	*,
	process_root: Path = Path("/proc"),
) -> bool:
	"""Detect a managed Wine game that outlived its launcher supervisor."""

	prefixes = {_normalized_path(prefix) for prefix in wine_prefixes}
	prefixes.discard("")
	if not prefixes:
		return False
	runtime_marker = _normalized_path(runtime_view).casefold().replace("\\", "/")
	if not runtime_marker:
		return False
	runtime_marker = runtime_marker.rstrip("/") + "/"
	try:
		processes = tuple(process_root.iterdir())
	except OSError:
		return False
	for process in processes:
		if not process.name.isdecimal():
			continue
		try:
			name = (process / "comm").read_text(
				encoding="utf-8",
				errors="replace",
			).strip().casefold()
		except OSError:
			continue
		if not name.endswith(".exe"):
			continue
		environment = _read_bounded(process / "environ", MAX_PROCESS_ENVIRONMENT_BYTES)
		if environment is None:
			continue
		wine_prefix = next(
			(
				entry.removeprefix(b"WINEPREFIX=")
				for entry in environment.split(b"\0")
				if entry.startswith(b"WINEPREFIX=")
			),
			None,
		)
		if (
			wine_prefix is None
			or _normalized_path(os.fsdecode(wine_prefix)) not in prefixes
		):
			continue
		command = _read_bounded(process / "cmdline", MAX_PROCESS_COMMAND_BYTES)
		if command is None:
			continue
		normalized_command = os.fsdecode(command).casefold().replace("\\", "/")
		if runtime_marker in normalized_command:
			return True
	return False


def _read_bounded(path: Path, maximum_bytes: int) -> bytes | None:
	try:
		with path.open("rb") as handle:
			value = handle.read(maximum_bytes + 1)
	except OSError:
		return None
	return value if len(value) <= maximum_bytes else None


def _normalized_path(path: Path | str) -> str:
	value = str(path)
	if not value:
		return ""
	return str(Path(value).expanduser().resolve(strict=False))


def _read_owner(handle: TextIO) -> str:
	handle.seek(0)
	return handle.read().strip() or "unknown operation"


def _write_owner(handle: TextIO, operation: str) -> None:
	handle.seek(0)
	handle.truncate()
	handle.write(
		f"{operation}, pid={os.getpid()}, "
		f"started={datetime.now().astimezone().isoformat()}\n"
	)
	handle.flush()