"""Plugin registry for tracking installed and enabled plugins.""" import json import logging from pathlib import Path from typing import Dict, List, Optional from dataclasses import dataclass, field, asdict from datetime import datetime, timezone logger = logging.getLogger(__name__) @dataclass class PluginState: """Persistent state for a plugin.""" name: str version: str installed_at: str enabled: bool = True migrations_applied: List[str] = field(default_factory=list) config: Dict = field(default_factory=dict) class PluginRegistry: """ Manages plugin state persistence. Stores state in JSON file in instance folder. """ def __init__(self, state_file: Path): self.state_file = state_file self._plugins: Dict[str, PluginState] = {} self._load() def _load(self) -> None: """Replace this instance's view with what is on disk. REPLACE, not merge. Assigning into the existing dict left entries that the file no longer has, so a plugin uninstalled through another worker stayed 'installed' here forever: enable_plugin would then skip install_plugin (and its migrations) and _save would write the ghost back out, resurrecting a plugin whose tables were never created. refresh() exists precisely to end stale reads, and it could not while this merged. """ loaded = {} if self.state_file.exists(): try: with open(self.state_file, 'r') as f: data = json.load(f) except (json.JSONDecodeError, OSError): data = None for name, state_data in ((data or {}).get('plugins') or {}).items(): try: loaded[name] = PluginState(**state_data) except TypeError: # A key written by a newer build. Skip THIS entry and keep # the rest: discarding the whole registry made every # subsequent enable a silent no-op, because enable() only # acts on names it can see and still answered 200. logger.warning( 'plugins.json entry %r has unexpected fields; ignoring ' 'it in this process', name) continue self._plugins = loaded self._migrate_renamed_plugins() def _migrate_renamed_plugins(self) -> None: """Upgrade path for the equipment -> machines plugin rename. Existing installs carry an 'equipment' entry in plugins.json. When the renamed plugins/machines directory exists and no 'machines' entry does, carry the state over under the new name and persist. """ if 'equipment' not in self._plugins or 'machines' in self._plugins: return machines_dir = Path(__file__).resolve().parents[2] / 'plugins' / 'machines' if not machines_dir.exists(): return state = self._plugins.pop('equipment') state.name = 'machines' self._plugins['machines'] = state self._save(only={'machines'}, remove={'equipment'}) def _save(self, remove: set = None, only: set = None) -> None: """Persist this registry, merging rather than overwriting. LOST UPDATES, AND WHY A PLAIN WRITE IS WRONG. Every gunicorn worker builds its own PluginRegistry and loads plugins.json once at boot, so each holds a snapshot that goes stale the moment another worker changes anything. Rewriting the whole file from that snapshot therefore erases whatever the other workers recorded: enabling seven plugins across four workers persisted three, and the API answered 200 for all seven. Silent data loss, with no error anywhere to hint at it. So take an exclusive lock, re-read what is on disk, apply only the entries this instance actually holds on top of it, and replace the file atomically. Concurrent writers then serialise instead of clobbering, and a crash mid-write cannot leave a half-written registry. """ import os import tempfile self.state_file.parent.mkdir(parents=True, exist_ok=True) lock_path = self.state_file.with_suffix('.lock') try: import fcntl except ImportError: # non-POSIX: fall back to the plain write fcntl = None lock_handle = None try: if fcntl is not None: lock_handle = open(lock_path, 'w') fcntl.flock(lock_handle, fcntl.LOCK_EX) merged = {} if self.state_file.exists(): try: with open(self.state_file) as handle: merged = (json.load(handle) or {}).get('plugins', {}) or {} except (json.JSONDecodeError, TypeError, OSError): merged = {} # Overlay ONLY the entries this call is changing. Writing every # entry this instance holds is what loses other workers' updates: # a snapshot taken at boot still says X is disabled, so saving an # unrelated change to Y drags X back down with it. `only` is the # mutation actually being persisted. for name, state in self._plugins.items(): if only is not None and name not in only: continue merged[name] = asdict(state) # A delete cannot be expressed by merging - the entry is simply # absent from self._plugins, and the on-disk copy would survive and # read as a resurrection. Name it instead. for name in (remove or set()): merged.pop(name, None) # Adopt the merged view so this instance stops carrying a stale # snapshot that a later save would write back out. rebuilt = {} for name, data in merged.items(): try: rebuilt[name] = PluginState(**data) except TypeError: # An unknown key from a newer build must not turn every # enable into a 500; keep the row on disk, skip it here. continue self._plugins = rebuilt handle = tempfile.NamedTemporaryFile( 'w', dir=str(self.state_file.parent), delete=False) try: json.dump({'plugins': merged}, handle, indent=2) handle.flush() os.fsync(handle.fileno()) finally: handle.close() os.replace(handle.name, self.state_file) return finally: if lock_handle is not None: try: fcntl.flock(lock_handle, fcntl.LOCK_UN) lock_handle.close() except Exception: pass def _save_unmerged(self) -> None: """Original whole-file write. Kept for callers that own the file.""" self.state_file.parent.mkdir(parents=True, exist_ok=True) with open(self.state_file, 'w') as f: json.dump({ 'plugins': { name: asdict(state) for name, state in self._plugins.items() } }, f, indent=2) def register(self, name: str, version: str, enabled: bool = True) -> PluginState: """Register a newly installed plugin. enabled=False leaves it off until a site opts in (used for plugins that provision extra tables).""" state = PluginState( name=name, version=version, installed_at=datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), enabled=enabled ) self._load() # another worker may have changed the file self._plugins[name] = state self._save(only={name}) return state def unregister(self, name: str) -> None: """Remove plugin from registry.""" self._load() # another worker may have registered it since boot self._plugins.pop(name, None) self._save(remove={name}) def refresh(self) -> None: """Re-read plugins.json before making a decision from it. Every worker loads this file once at boot, so any read used to DECIDE something - is it installed, is a dependency enabled - is answering from a snapshot that another worker may have moved on from. That is how enabling endpoints returned 409 "computers is not enabled" moments after computers had in fact been enabled through a different worker. """ self._load() def get(self, name: str) -> Optional[PluginState]: """Get plugin state.""" return self._plugins.get(name) def is_installed(self, name: str) -> bool: """Check if plugin is installed.""" return name in self._plugins def is_enabled(self, name: str) -> bool: """Check if plugin is enabled.""" state = self._plugins.get(name) return state.enabled if state else False def enable(self, name: str) -> None: """Enable a plugin. Re-reads first: this worker's snapshot dates from boot, so a plugin another worker installed since is absent here and the old `if name in self._plugins` guard silently did nothing. """ self._load() if name in self._plugins: self._plugins[name].enabled = True self._save(only={name}) def disable(self, name: str) -> None: """Disable a plugin. Re-reads first, for the same reason as enable.""" self._load() if name in self._plugins: self._plugins[name].enabled = False self._save(only={name}) def get_enabled_plugins(self) -> List[str]: """Get list of enabled plugin names.""" return [ name for name, state in self._plugins.items() if state.enabled ] def add_migration(self, name: str, revision: str) -> None: """Record that a migration was applied.""" self._load() if name in self._plugins: if revision not in self._plugins[name].migrations_applied: self._plugins[name].migrations_applied.append(revision) self._save(only={name}) def get_all(self) -> Dict[str, PluginState]: """Get all registered plugins.""" return self._plugins.copy() def update_config(self, name: str, config: Dict) -> None: """Update plugin configuration.""" self._load() if name in self._plugins: self._plugins[name].config.update(config) self._save(only={name})