Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 43 additions & 4 deletions src/cache_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,21 @@

class CacheManager:
"""Manages caching of API responses to reduce API calls."""


# Which cache directories already have a cleanup thread in this process.
#
# The sweep is directory-scoped work -- it lists a directory and deletes
# from it -- so one per directory is the right number no matter how many
# managers exist. Nothing enforced that before: every instance started its
# own, and because the loop closes over `self`, a discarded manager could
# never be collected and its thread woke to re-scan the same directory
# every 24 hours for the life of the process. Startup validation runs
# twice and built a throwaway manager each time, so a display process
# carried three threads for one cache.
_cleanup_owners: Dict[str, 'CacheManager'] = {}
_cleanup_owners_lock = threading.Lock()


def __init__(self) -> None:
# Initialize logger first
self.logger: logging.Logger = get_logger(__name__)
Expand Down Expand Up @@ -718,11 +732,29 @@ def cleanup_disk_cache(self, force: bool = False) -> Dict[str, Any]:
}

def start_cleanup_thread(self) -> None:
"""Start background thread for periodic disk cache cleanup."""
"""Start background thread for periodic disk cache cleanup.

At most one thread per cache directory per process: the sweep is
directory-scoped, so a second one only duplicates the scan.
"""
if self._cleanup_thread and self._cleanup_thread.is_alive():
self.logger.debug("Cleanup thread already running")
return


with CacheManager._cleanup_owners_lock:
owner = CacheManager._cleanup_owners.get(self.cache_dir)
if owner is not None and owner is not self:
thread = owner._cleanup_thread
if thread is not None and thread.is_alive():
self.logger.debug(
"Cleanup thread for %s already owned by another cache "
"manager in this process; not starting a second",
self.cache_dir)
return
# The owner's thread died or was stopped -- take over.
CacheManager._cleanup_owners[self.cache_dir] = self


def cleanup_loop():
"""Background loop that runs cleanup periodically."""
self.logger.info("Disk cache cleanup thread started (interval: %d hours)",
Expand Down Expand Up @@ -770,10 +802,17 @@ def stop_cleanup_thread(self) -> None:
Signals the thread to stop and waits for it to finish (with timeout).
This allows for clean shutdown during testing or application termination.
"""
# Release ownership first and unconditionally, so a manager that never
# started a thread (or whose thread already exited) cannot keep the
# directory claimed and block a live manager from sweeping it.
with CacheManager._cleanup_owners_lock:
if CacheManager._cleanup_owners.get(self.cache_dir) is self:
del CacheManager._cleanup_owners[self.cache_dir]

if not self._cleanup_thread or not self._cleanup_thread.is_alive():
self.logger.debug("Cleanup thread not running")
return

self.logger.info("Stopping disk cache cleanup thread...")
self._cleanup_stop_event.set() # Signal thread to stop

Expand Down
6 changes: 4 additions & 2 deletions src/display_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,8 @@ def __init__(self):
# Validate startup configuration
try:
from src.startup_validator import StartupValidator
validator = StartupValidator(self.config_manager)
validator = StartupValidator(self.config_manager,
cache_manager=self.cache_manager)
is_valid, errors, warnings = validator.validate_all()

if warnings:
Expand Down Expand Up @@ -258,7 +259,8 @@ def _follower_gated_update():
# Validate plugins after plugin manager is created
try:
from src.startup_validator import StartupValidator
validator = StartupValidator(self.config_manager, self.plugin_manager)
validator = StartupValidator(self.config_manager, self.plugin_manager,
cache_manager=self.cache_manager)
is_valid, errors, warnings = validator.validate_all()

if warnings:
Expand Down
29 changes: 24 additions & 5 deletions src/startup_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,16 +15,23 @@
class StartupValidator:
"""Validates system state on startup."""

def __init__(self, config_manager: Any, plugin_manager: Optional[Any] = None) -> None:
def __init__(self, config_manager: Any, plugin_manager: Optional[Any] = None,
cache_manager: Optional[Any] = None) -> None:
"""
Initialize the startup validator.

Args:
config_manager: ConfigManager instance
plugin_manager: Optional PluginManager instance
cache_manager: The CacheManager the application will actually use.
Pass it. Without one this validator builds its own just to read
a directory path, which reports on a cache the app does not
use and leaves behind a cleanup thread that nothing stops --
validation runs twice per startup, so that was two of them.
"""
self.config_manager = config_manager
self.plugin_manager = plugin_manager
self.cache_manager = cache_manager
self.logger = get_logger(__name__)
self.errors: List[str] = []
self.warnings: List[str] = []
Expand Down Expand Up @@ -91,9 +98,21 @@ def _validate_config(self) -> None:
def _validate_cache_directory(self) -> None:
"""Validate cache directory permissions."""
try:
from src.cache_manager import CacheManager
cache_manager = CacheManager()
cache_dir = cache_manager.get_cache_dir()
cache_manager = self.cache_manager
if cache_manager is None:
# No caller supplied one (older embedders, direct use in a
# script). Build one, but do not leave its cleanup thread
# running behind us -- this instance is discarded on the next
# line but the thread is a closure over it, so it would never
# be collected.
from src.cache_manager import CacheManager
cache_manager = CacheManager()
try:
cache_dir = cache_manager.get_cache_dir()
finally:
cache_manager.stop_cleanup_thread()
else:
cache_dir = cache_manager.get_cache_dir()

if not cache_dir:
self.warnings.append("Cache directory not available - caching will be disabled")
Expand Down
146 changes: 146 additions & 0 deletions test/test_cache_cleanup_thread_ownership.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
"""Tests that one cache directory gets one cleanup thread per process.

The sweep lists a directory and deletes from it, so a second thread over the
same directory only duplicates the scan. Nothing enforced that: every
CacheManager started its own, and since the loop closes over `self`, a
discarded manager could never be collected -- its thread stayed alive and
re-scanned the same directory every 24 hours for the life of the process.

On the dev rig a display process carried three, for one cache directory:

14:22:59.954 display_controller (the real one)
14:22:59.973 startup validation, run 1 (discarded)
14:23:01.055 startup validation, run 2 (discarded)

Startup validation runs twice and built a throwaway manager each time, purely
to read a directory path.
"""

import threading

import pytest

from src.cache_manager import CacheManager


@pytest.fixture(autouse=True)
def _clean_registry():
CacheManager._cleanup_owners.clear()
yield
for owner in list(CacheManager._cleanup_owners.values()):
owner.stop_cleanup_thread()
CacheManager._cleanup_owners.clear()


def _live_cleanup_threads():
return [t for t in threading.enumerate()
if t.name == 'DiskCacheCleanup' and t.is_alive()]


@pytest.fixture
def manager(tmp_path, monkeypatch):
"""A CacheManager pinned to a temp dir, so tests never touch the real one."""
monkeypatch.setattr(CacheManager, '_get_writable_cache_dir',
lambda self: str(tmp_path))
return CacheManager


class TestOneThreadPerDirectory:
def test_a_single_manager_starts_one(self, manager):
before = len(_live_cleanup_threads())
m = manager()
try:
assert len(_live_cleanup_threads()) == before + 1
finally:
m.stop_cleanup_thread()

def test_three_managers_still_start_one(self, manager):
# Exactly the rig's shape: the real manager plus two throwaways.
before = len(_live_cleanup_threads())
managers = [manager() for _ in range(3)]
try:
assert len(_live_cleanup_threads()) == before + 1
finally:
for m in managers:
m.stop_cleanup_thread()

def test_the_first_one_owns_it(self, manager):
first, second = manager(), manager()
try:
assert CacheManager._cleanup_owners[first.cache_dir] is first
assert second._cleanup_thread is None
finally:
first.stop_cleanup_thread()
second.stop_cleanup_thread()

def test_the_survivor_can_take_over(self, manager):
first = manager()
first.stop_cleanup_thread()
assert not _live_cleanup_threads()

second = manager()
try:
# Ownership was released, so the directory is swept again rather
# than being left permanently unclaimed by a dead owner.
assert len(_live_cleanup_threads()) == 1
assert CacheManager._cleanup_owners[second.cache_dir] is second
finally:
second.stop_cleanup_thread()

def test_stopping_a_non_owner_does_not_unclaim_the_directory(self, manager):
first, second = manager(), manager()
try:
second.stop_cleanup_thread() # never owned it
assert CacheManager._cleanup_owners[first.cache_dir] is first
assert len(_live_cleanup_threads()) == 1
finally:
first.stop_cleanup_thread()

def test_separate_directories_get_separate_threads(self, tmp_path, monkeypatch):
a, b = tmp_path / 'a', tmp_path / 'b'
a.mkdir()
b.mkdir()
dirs = iter([str(a), str(b)])
monkeypatch.setattr(CacheManager, '_get_writable_cache_dir',
lambda self: next(dirs))
first, second = CacheManager(), CacheManager()
try:
assert first.cache_dir != second.cache_dir
assert len(_live_cleanup_threads()) == 2
finally:
first.stop_cleanup_thread()
second.stop_cleanup_thread()

def test_no_thread_leaks_across_many_constructions(self, manager):
before = len(_live_cleanup_threads())
made = [manager() for _ in range(12)]
try:
assert len(_live_cleanup_threads()) == before + 1
finally:
for m in made:
m.stop_cleanup_thread()
assert len(_live_cleanup_threads()) == before


class TestValidatorDoesNotBuildItsOwn:
def test_it_uses_the_cache_manager_it_is_given(self, manager):
from src.startup_validator import StartupValidator

shared = manager()
try:
before = len(_live_cleanup_threads())
v = StartupValidator(config_manager=object(), cache_manager=shared)
v._validate_cache_directory()
assert len(_live_cleanup_threads()) == before, (
"validation started another cleanup thread")
finally:
shared.stop_cleanup_thread()

def test_without_one_it_cleans_up_after_itself(self, manager):
from src.startup_validator import StartupValidator

before = len(_live_cleanup_threads())
v = StartupValidator(config_manager=object())
v._validate_cache_directory()
assert len(_live_cleanup_threads()) == before, (
"the fallback manager left its cleanup thread running")
Loading