From 08a4c3b80402f858341619ee27a39521be85af51 Mon Sep 17 00:00:00 2001 From: Katie Strader Date: Wed, 1 Jul 2026 13:52:59 -0500 Subject: [PATCH 1/3] fix: added retry to pipeline.py to mitigate WinError 5 "access denied", shared RotatingFileHandler cache so root and dlt loggers reuse one handler per path, avoiding rotation "file in use" errors --- src/openhound/core/logging.py | 77 ++++++++++------------- src/openhound/core/pipeline.py | 93 +++++++++++++++++++++------- tests/test_log_handlers.py | 106 ++++++++++++++++++++++++++++++++ tests/test_pipeline_retry.py | 108 +++++++++++++++++++++++++++++++++ 4 files changed, 316 insertions(+), 68 deletions(-) create mode 100644 tests/test_pipeline_retry.py diff --git a/src/openhound/core/logging.py b/src/openhound/core/logging.py index 2582719..ba219fe 100644 --- a/src/openhound/core/logging.py +++ b/src/openhound/core/logging.py @@ -238,6 +238,10 @@ def __init__( self.base_path = Path(base_path) if base_path else self.default_platform_path() self.log_file_path: Path | None = None + # Share one rotating handler per file across loggers; opening the same file + # with multiple handlers breaks rotation on Windows (WinError 32). + self._file_handlers: dict[Path, "RotatingFileHandler"] = {} + self.handlers = { LogMode.CLI: self.cli_handlers, LogMode.CONTAINER: self.container_handlers, @@ -370,18 +374,8 @@ def _file_formatter(self) -> logging.Formatter: return OpenHoundJSONFormatter() return OpenHoundTextFormatter() - def container_handlers(self, logger: logging.Logger, file_path: Path) -> None: - """Set the logging handler/format when running in a container""" - - formatter = self._file_formatter() - - # Log to stdout for better compatibility with container-based logging systems. - # Output is human-readable text by default; set runtime.log_format = "JSON" for structured JSON. - stdout_handler = logging.StreamHandler(sys.stdout) - stdout_handler.setFormatter(formatter) - logger.addHandler(stdout_handler) - - # But also log the same format to a file for persistence and debugging when needed + def _build_file_handler(self, file_path: Path) -> "RotatingFileHandler": + """Create and configure a rotating file handler for the given path.""" rotating_file_handler = RotatingFileHandler( file_path, when=self.rotate_when, @@ -389,14 +383,36 @@ def container_handlers(self, logger: logging.Logger, file_path: Path) -> None: backupCount=self.backup_count, max_bytes=self.max_bytes, ) - rotating_file_handler.setFormatter(formatter) + rotating_file_handler.setFormatter(self._file_formatter()) # This regular expression overrides the default extMatch to recognize both # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well) rotating_file_handler.extMatch = re.compile( r"(? "RotatingFileHandler": + """Return a shared rotating file handler for the path (one open file per path).""" + key = Path(file_path).resolve() + handler = self._file_handlers.get(key) + if handler is None: + handler = self._build_file_handler(file_path) + self._file_handlers[key] = handler + return handler + + def container_handlers(self, logger: logging.Logger, file_path: Path) -> None: + """Set the logging handler/format when running in a container""" + + formatter = self._file_formatter() + + # Log to stdout for better compatibility with container-based logging systems. + # Output is human-readable text by default; set runtime.log_format = "JSON" for structured JSON. + stdout_handler = logging.StreamHandler(sys.stdout) + stdout_handler.setFormatter(formatter) + logger.addHandler(stdout_handler) + + # But also log the same format to a file for persistence and debugging when needed + logger.addHandler(self._get_file_handler(file_path)) def cli_handlers(self, logger: logging.Logger, file_path: Path) -> None: """Set the logging handler/format when running as a standalone CLI tool""" @@ -416,40 +432,11 @@ def cli_handlers(self, logger: logging.Logger, file_path: Path) -> None: logger.addHandler(console_handler) # But also save the logs to a file using the configured format (text by default) - file_formatter = self._file_formatter() - rotating_file_handler = RotatingFileHandler( - file_path, - when=self.rotate_when, - interval=self.interval, - backupCount=self.backup_count, - max_bytes=self.max_bytes, - ) - rotating_file_handler.setFormatter(file_formatter) - # This regular expression overrides the default extMatch to recognize both - # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well) - rotating_file_handler.extMatch = re.compile( - r"(? None: """Set the logging handler/format when running the OpenHound service""" - file_formatter = self._file_formatter() - rotating_file_handler = RotatingFileHandler( - file_path, - when=self.rotate_when, - interval=self.interval, - backupCount=self.backup_count, - max_bytes=self.max_bytes, - ) - rotating_file_handler.setFormatter(file_formatter) - # This regular expression overrides the default extMatch to recognize both - # default time based rotation filenames and size based rotation filenames (which gets a seconds added as well) - rotating_file_handler.extMatch = re.compile( - r"(? LogMode: diff --git a/src/openhound/core/pipeline.py b/src/openhound/core/pipeline.py index 6399195..deddc8c 100644 --- a/src/openhound/core/pipeline.py +++ b/src/openhound/core/pipeline.py @@ -1,3 +1,6 @@ +import errno +import logging +import time from abc import ABC, abstractmethod from dlt.common.configuration.exceptions import ConfigFieldMissingException @@ -8,6 +11,27 @@ from openhound.core.exceptions import ConfigException, ParseException +logger = logging.getLogger(__name__) + +# Windows can transiently lock freshly written load-package files, so dlt's +# per-file rename fallback raises PermissionError (WinError 5); retry the run. +_TRANSIENT_RENAME_ERRNOS = {errno.EACCES, errno.EPERM} +_MAX_TRANSIENT_RETRIES = 5 +_TRANSIENT_RETRY_BACKOFF = 0.25 + + +def _transient_filesystem_cause(err: BaseException) -> OSError | None: + """Return the underlying transient filesystem error in the cause chain, if any.""" + seen: set[int] = set() + current: BaseException | None = err + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, OSError) and current.errno in _TRANSIENT_RENAME_ERRNOS: + return current + nested = getattr(current, "exception", None) + current = nested if nested is not None else current.__cause__ + return None + class BasePipeline(ABC): @property @@ -15,26 +39,49 @@ class BasePipeline(ABC): def pipeline(self) -> "Pipeline": ... def _run(self, source, **kwargs) -> LoadInfo: - try: - return self.pipeline.run(source, **kwargs) - except PipelineStepFailed as err: - if isinstance(err.exception, ConfigFieldMissingException): - config_cause: ConfigFieldMissingException = err.exception - raise ConfigException( - pipeline_name=err.pipeline.pipeline_name, - destination=str(err.pipeline.destination), - dataset_name=err.pipeline.dataset_name, - spec_name=config_cause.spec_name, - message=str(err.exception), - ) from None - - if isinstance(err.exception, ResourceExtractionError): - extract_cause: ResourceExtractionError = err.exception - raise ParseException( - pipeline_name=err.pipeline.pipeline_name, - destination=str(err.pipeline.destination), - dataset_name=err.pipeline.dataset_name, - step=extract_cause.pipe_name, - message=extract_cause.msg, - ) - raise + last_err: BaseException | None = None + for attempt in range(_MAX_TRANSIENT_RETRIES): + try: + return self.pipeline.run(source, **kwargs) + except PipelineStepFailed as err: + if isinstance(err.exception, ConfigFieldMissingException): + config_cause: ConfigFieldMissingException = err.exception + raise ConfigException( + pipeline_name=err.pipeline.pipeline_name, + destination=str(err.pipeline.destination), + dataset_name=err.pipeline.dataset_name, + spec_name=config_cause.spec_name, + message=str(err.exception), + ) from None + + if isinstance(err.exception, ResourceExtractionError): + extract_cause: ResourceExtractionError = err.exception + raise ParseException( + pipeline_name=err.pipeline.pipeline_name, + destination=str(err.pipeline.destination), + dataset_name=err.pipeline.dataset_name, + step=extract_cause.pipe_name, + message=extract_cause.msg, + ) + + if _transient_filesystem_cause(err) is None: + raise + last_err = err + except PermissionError as err: + if err.errno not in _TRANSIENT_RENAME_ERRNOS: + raise + last_err = err + + if attempt + 1 >= _MAX_TRANSIENT_RETRIES: + break + + logger.warning( + "Transient filesystem error; retrying run (%d/%d): %s", + attempt + 1, + _MAX_TRANSIENT_RETRIES, + last_err, + ) + time.sleep(_TRANSIENT_RETRY_BACKOFF * (attempt + 1)) + + assert last_err is not None + raise last_err diff --git a/tests/test_log_handlers.py b/tests/test_log_handlers.py index 1a6ca1f..bb7b7bb 100644 --- a/tests/test_log_handlers.py +++ b/tests/test_log_handlers.py @@ -1,5 +1,6 @@ import json import logging +from pathlib import Path import pytest @@ -156,3 +157,108 @@ def test_text_formatter_produces_plain_text(): with pytest.raises(json.JSONDecodeError): json.loads(output) + + +def test_get_file_handler_caches_handler_per_path(tmp_path): + """The shared handler cache should return the same instance for a given path and + distinct instances for different paths, so each file is only opened once.""" + custom_logger = CustomLogger("openhound.log", base_path=str(tmp_path)) + + path_a = tmp_path / "openhound.log" + path_b = tmp_path / "ext_test.log" + + handler_a = custom_logger._get_file_handler(path_a) + try: + # A repeated lookup (even via a re-built equivalent Path) returns the cache hit + assert custom_logger._get_file_handler(path_a) is handler_a, ( + "The same path should return the cached handler instance" + ) + assert custom_logger._get_file_handler(Path(str(path_a))) is handler_a, ( + "Equivalent paths should resolve to the same cached handler" + ) + + # A different path gets its own dedicated handler + handler_b = custom_logger._get_file_handler(path_b) + assert handler_b is not handler_a, ( + "A different path should produce a different handler instance" + ) + assert isinstance(handler_b, RotatingFileHandler), ( + "Cached handlers should be RotatingFileHandler instances" + ) + finally: + for handler in custom_logger._file_handlers.values(): + handler.close() + + +def test_root_and_dlt_loggers_share_single_file_handler(tmp_path, monkeypatch): + """The root and dlt loggers writing to openhound.log must share one handler + instance so the file is only opened once, which is required for rotation on + Windows where an open handle blocks renaming the file.""" + # A sibling module may set RUNTIME__LOG_PATH on import; keep our explicit base_path. + monkeypatch.delenv("RUNTIME__LOG_PATH", raising=False) + custom_logger = CustomLogger("openhound.log", base_path=str(tmp_path)) + + try: + custom_logger.setup() + + root_logger = logging.getLogger() + dlt_logger = logging.getLogger("dlt") + + root_file_handlers = [ + handler + for handler in root_logger.handlers + if isinstance(handler, RotatingFileHandler) + ] + dlt_file_handlers = [ + handler + for handler in dlt_logger.handlers + if isinstance(handler, RotatingFileHandler) + ] + + assert len(root_file_handlers) == 1, ( + "The root logger should have exactly one rotating file handler" + ) + assert len(dlt_file_handlers) == 1, ( + "The dlt logger should have exactly one rotating file handler" + ) + assert root_file_handlers[0] is dlt_file_handlers[0], ( + "Root and dlt loggers must share the same handler instance for openhound.log" + ) + assert root_file_handlers[0].baseFilename.endswith("openhound.log"), ( + "The shared handler should write to 'openhound.log'" + ) + finally: + # Restore the shared global logging state for subsequent tests. + logger_override.setup() + + +def test_build_file_handler_applies_rotation_settings(tmp_path): + """_build_file_handler should produce a RotatingFileHandler configured with the + logger's rotation settings, the custom extMatch, and the selected formatter.""" + custom_logger = CustomLogger( + "openhound.log", + base_path=str(tmp_path), + backup_count=7, + max_bytes=1234, + log_format="json", + ) + + handler = custom_logger._build_file_handler(tmp_path / "openhound.log") + try: + assert isinstance(handler, RotatingFileHandler), ( + "The built handler should be a RotatingFileHandler" + ) + assert handler.backupCount == 7, "The configured backup_count should be applied" + assert handler.max_bytes == 1234, "The configured max_bytes should be applied" + assert isinstance(handler.formatter, OpenHoundJSONFormatter), ( + "log_format 'json' should attach the JSON formatter to the handler" + ) + # The overridden extMatch must recognize both time- and size-based suffixes + assert handler.extMatch.match("2024-01-02_03"), ( + "extMatch should recognize the time-based rotation suffix" + ) + assert handler.extMatch.match("2024-01-02_03-04-05"), ( + "extMatch should recognize the size-based rotation suffix" + ) + finally: + handler.close() diff --git a/tests/test_pipeline_retry.py b/tests/test_pipeline_retry.py new file mode 100644 index 0000000..4ca801b --- /dev/null +++ b/tests/test_pipeline_retry.py @@ -0,0 +1,108 @@ +import errno + +import pytest +from dlt.pipeline.exceptions import PipelineStepFailed + +from openhound.core.pipeline import _MAX_TRANSIENT_RETRIES, BasePipeline + + +class _DummyDltPipeline: + pipeline_name = "test_pipeline" + has_pending_data = True + + +def _wrap_in_step_failed(exc: BaseException) -> PipelineStepFailed: + """Wrap an exception the way dlt does when a load/normalize step fails.""" + return PipelineStepFailed( + pipeline=_DummyDltPipeline(), + step="load", + load_id="1782853056", + exception=exc, + ) + + +class _FakeDlt: + """Stand-in for a dlt Pipeline whose run() replays a list of side effects.""" + + def __init__(self, side_effects): + self._side_effects = list(side_effects) + self.calls = 0 + + def run(self, source, **kwargs): + self.calls += 1 + effect = self._side_effects.pop(0) + if isinstance(effect, BaseException): + raise effect + return effect + + +class _FakePipeline(BasePipeline): + def __init__(self, fake_dlt: _FakeDlt): + self._fake = fake_dlt + + @property + def pipeline(self): + return self._fake + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + monkeypatch.setattr("openhound.core.pipeline.time.sleep", lambda *_: None) + + +def _winerror5() -> PermissionError: + return PermissionError(errno.EACCES, "Access is denied") + + +def test_run_retries_transient_permission_error_wrapped(): + sentinel = object() + fake = _FakeDlt( + [ + _wrap_in_step_failed(_winerror5()), + _wrap_in_step_failed(_winerror5()), + sentinel, + ] + ) + pipeline = _FakePipeline(fake) + + assert pipeline._run(source=None) is sentinel + assert fake.calls == 3 + + +def test_run_retries_bare_permission_error(): + sentinel = object() + fake = _FakeDlt([_winerror5(), sentinel]) + pipeline = _FakePipeline(fake) + + assert pipeline._run(source=None) is sentinel + assert fake.calls == 2 + + +def test_run_reraises_after_max_retries(): + fake = _FakeDlt([_wrap_in_step_failed(_winerror5())] * _MAX_TRANSIENT_RETRIES) + pipeline = _FakePipeline(fake) + + with pytest.raises(PipelineStepFailed): + pipeline._run(source=None) + + assert fake.calls == _MAX_TRANSIENT_RETRIES + + +def test_run_does_not_retry_non_transient_permission_error(): + fake = _FakeDlt([PermissionError(errno.ENOTEMPTY, "Directory not empty")]) + pipeline = _FakePipeline(fake) + + with pytest.raises(PermissionError): + pipeline._run(source=None) + + assert fake.calls == 1 + + +def test_run_does_not_retry_other_pipeline_step_failures(): + fake = _FakeDlt([_wrap_in_step_failed(RuntimeError("boom"))]) + pipeline = _FakePipeline(fake) + + with pytest.raises(PipelineStepFailed): + pipeline._run(source=None) + + assert fake.calls == 1 From 2a891dbba01da0221bf66d7121f6a04f122c04d8 Mon Sep 17 00:00:00 2001 From: Katie Strader Date: Wed, 1 Jul 2026 15:54:17 -0500 Subject: [PATCH 2/3] chore: instead of asser, raise RuntimeError --- src/openhound/core/pipeline.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/openhound/core/pipeline.py b/src/openhound/core/pipeline.py index deddc8c..825b2ee 100644 --- a/src/openhound/core/pipeline.py +++ b/src/openhound/core/pipeline.py @@ -83,5 +83,6 @@ def _run(self, source, **kwargs) -> LoadInfo: ) time.sleep(_TRANSIENT_RETRY_BACKOFF * (attempt + 1)) - assert last_err is not None + if last_err is None: + raise RuntimeError("unreachable: retry loop exited without an error") raise last_err From 2edc1223d43d010f28dd9bd016da09bf217fdd01 Mon Sep 17 00:00:00 2001 From: Stran Dutton Date: Wed, 1 Jul 2026 16:24:18 -0500 Subject: [PATCH 3/3] improve error surfacing, platform-gate the permission erorr to Windows, testing --- src/openhound/core/pipeline.py | 7 +++++-- tests/test_pipeline_retry.py | 32 +++++++++++++++++++++++++++++++- 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/src/openhound/core/pipeline.py b/src/openhound/core/pipeline.py index 825b2ee..0aaeb9d 100644 --- a/src/openhound/core/pipeline.py +++ b/src/openhound/core/pipeline.py @@ -1,5 +1,6 @@ import errno import logging +import sys import time from abc import ABC, abstractmethod @@ -29,7 +30,9 @@ def _transient_filesystem_cause(err: BaseException) -> OSError | None: if isinstance(current, OSError) and current.errno in _TRANSIENT_RENAME_ERRNOS: return current nested = getattr(current, "exception", None) - current = nested if nested is not None else current.__cause__ + if nested is None: + nested = current.__cause__ or current.__context__ + current = nested return None @@ -68,7 +71,7 @@ def _run(self, source, **kwargs) -> LoadInfo: raise last_err = err except PermissionError as err: - if err.errno not in _TRANSIENT_RENAME_ERRNOS: + if sys.platform != "win32" or err.errno not in _TRANSIENT_RENAME_ERRNOS: raise last_err = err diff --git a/tests/test_pipeline_retry.py b/tests/test_pipeline_retry.py index 4ca801b..69cab08 100644 --- a/tests/test_pipeline_retry.py +++ b/tests/test_pipeline_retry.py @@ -69,7 +69,8 @@ def test_run_retries_transient_permission_error_wrapped(): assert fake.calls == 3 -def test_run_retries_bare_permission_error(): +def test_run_retries_bare_permission_error(monkeypatch): + monkeypatch.setattr("openhound.core.pipeline.sys.platform", "win32") sentinel = object() fake = _FakeDlt([_winerror5(), sentinel]) pipeline = _FakePipeline(fake) @@ -78,6 +79,35 @@ def test_run_retries_bare_permission_error(): assert fake.calls == 2 +def test_run_does_not_retry_bare_permission_error_on_non_windows(monkeypatch): + monkeypatch.setattr("openhound.core.pipeline.sys.platform", "linux") + fake = _FakeDlt([_winerror5()]) + pipeline = _FakePipeline(fake) + + with pytest.raises(PermissionError): + pipeline._run(source=None) + + assert fake.calls == 1 + + +def test_run_retries_implicit_chained_permission_error(): + """PermissionError on __context__ (raised without 'from') should be detected and retried.""" + sentinel = object() + # Simulate: some intermediate error raised inside `except PermissionError:` (no 'from'), + # so __context__ is set to the PermissionError but __cause__ is not. + perm_err = _winerror5() + intermediate = RuntimeError("intermediate") + intermediate.__context__ = perm_err + # dlt wraps the intermediate error in PipelineStepFailed + step_failed = _wrap_in_step_failed(intermediate) + + fake = _FakeDlt([step_failed, sentinel]) + pipeline = _FakePipeline(fake) + + assert pipeline._run(source=None) is sentinel + assert fake.calls == 2 + + def test_run_reraises_after_max_retries(): fake = _FakeDlt([_wrap_in_step_failed(_winerror5())] * _MAX_TRANSIENT_RETRIES) pipeline = _FakePipeline(fake)