From 40b815e2b02a2d52910380b88636645e6c612130 Mon Sep 17 00:00:00 2001 From: GautamSharma99 Date: Tue, 14 Jul 2026 18:49:14 +0530 Subject: [PATCH] fix(server): offload blocking git operations --- .../src/repowise/server/job_executor.py | 22 +++-- .../server/src/repowise/server/scheduler.py | 5 +- tests/unit/server/test_job_executor.py | 82 +++++++++++++++++++ tests/unit/server/test_scheduler_async.py | 44 ++++++++++ 4 files changed, 142 insertions(+), 11 deletions(-) create mode 100644 tests/unit/server/test_scheduler_async.py diff --git a/packages/server/src/repowise/server/job_executor.py b/packages/server/src/repowise/server/job_executor.py index ce38fcd4e..34b62d577 100644 --- a/packages/server/src/repowise/server/job_executor.py +++ b/packages/server/src/repowise/server/job_executor.py @@ -484,7 +484,8 @@ async def execute_job( # mode, config) that `repowise init` would have written. try: if is_initial_index: - _persist_initial_index_state( + await asyncio.to_thread( + _persist_initial_index_state, Path(repo_path), llm_client=llm_client, docs_enabled=generate_docs, @@ -494,7 +495,7 @@ async def execute_job( exclude_patterns=exclude_patterns, ) else: - _stamp_last_sync_commit(Path(repo_path)) + await asyncio.to_thread(_stamp_last_sync_commit, Path(repo_path)) except Exception: logger.debug("state_json_update_failed", job_id=job_id, exc_info=True) @@ -739,7 +740,8 @@ async def _incremental_page_regen( import subprocess as _sp - head_result = _sp.run( + head_result = await asyncio.to_thread( + _sp.run, ["git", "rev-parse", "HEAD"], cwd=str(repo_path), capture_output=True, @@ -758,13 +760,17 @@ async def _incremental_page_regen( from repowise.core.ingestion.change_detector import compute_adaptive_budget detector = ChangeDetector(repo_path) - file_diffs = detector.get_changed_files(base_ref, head) + file_diffs = await asyncio.to_thread(detector.get_changed_files, base_ref, head) if not file_diffs: return [] cascade_budget = compute_adaptive_budget(file_diffs, result.file_count) - affected = detector.get_affected_pages( - file_diffs, result.graph_builder.graph(), cascade_budget + graph = await asyncio.to_thread(result.graph_builder.graph) + affected = await asyncio.to_thread( + detector.get_affected_pages, + file_diffs, + graph, + cascade_budget, ) if not affected.regenerate: @@ -806,9 +812,7 @@ async def _incremental_page_regen( language=repo_cfg.get("language", "en"), ) assembler = ContextAssembler(generation_config) - generator = PageGenerator( - llm_client, assembler, generation_config, repo_path=repo_path - ) + generator = PageGenerator(llm_client, assembler, generation_config, repo_path=repo_path) pages = await generator.generate_all( affected_parsed, diff --git a/packages/server/src/repowise/server/scheduler.py b/packages/server/src/repowise/server/scheduler.py index 18c608472..6f992d77d 100644 --- a/packages/server/src/repowise/server/scheduler.py +++ b/packages/server/src/repowise/server/scheduler.py @@ -107,7 +107,8 @@ async def polling_fallback() -> None: # Get current git HEAD try: - head_result = subprocess.run( + head_result = await asyncio.to_thread( + subprocess.run, ["git", "rev-parse", "HEAD"], cwd=repo.local_path, capture_output=True, @@ -134,7 +135,7 @@ async def polling_fallback() -> None: continue # Enqueue a sync job - await crud.upsert_generation_job( + job = await crud.upsert_generation_job( session, repository_id=repo.id, status="pending", diff --git a/tests/unit/server/test_job_executor.py b/tests/unit/server/test_job_executor.py index 5f422bf2d..f67f67463 100644 --- a/tests/unit/server/test_job_executor.py +++ b/tests/unit/server/test_job_executor.py @@ -8,6 +8,7 @@ from __future__ import annotations +import threading from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -228,6 +229,33 @@ async def test_execute_job_merges_config_yaml_excludes(session_factory, tmp_path assert run_pipeline_mock.await_args.kwargs["exclude_patterns"] == ["tools/"] +@pytest.mark.asyncio +async def test_execute_job_offloads_state_stamping(session_factory, tmp_path): + """The final Git HEAD lookup and state write should not block the event loop.""" + job_id = await _seed_repo_and_job(session_factory, tmp_path) + app_state = SimpleNamespace(session_factory=session_factory, fts=None, vector_store=None) + main_thread = threading.get_ident() + stamp_thread: int | None = None + + def stamp_state(repo_path): + nonlocal stamp_thread + stamp_thread = threading.get_ident() + + with ( + patch("repowise.server.job_executor.run_pipeline", AsyncMock(return_value=_fake_result())), + patch("repowise.server.job_executor.persist_pipeline_result", AsyncMock()), + patch("repowise.server.job_executor._stamp_last_sync_commit", side_effect=stamp_state), + patch( + "repowise.server.provider_config.get_chat_provider_instance", + side_effect=RuntimeError("no provider"), + ), + ): + await execute_job(job_id, app_state) + + assert stamp_thread is not None + assert stamp_thread != main_thread + + @pytest.mark.asyncio async def test_incremental_page_regen_passes_repo_path(tmp_path): """Incremental regen must forward repo_path to generate_all. @@ -289,3 +317,57 @@ async def test_incremental_page_regen_passes_repo_path(tmp_path): generator.generate_all.assert_awaited_once() assert generator.generate_all.await_args.kwargs["repo_path"] == Path(repo_path) + + +@pytest.mark.asyncio +async def test_incremental_page_regen_offloads_change_detection(tmp_path): + """Git and change detection should run outside the event-loop thread.""" + repo_path = tmp_path + repowise_dir = repo_path / ".repowise" + repowise_dir.mkdir() + (repowise_dir / "state.json").write_text('{"last_sync_commit": "base-sha"}', encoding="utf-8") + + main_thread = threading.get_ident() + worker_threads: dict[str, int] = {} + + def head_lookup(*args, **kwargs): + worker_threads["head"] = threading.get_ident() + return SimpleNamespace(returncode=0, stdout="head-sha\n") + + detector = MagicMock() + + def get_changed_files(*args): + worker_threads["diff"] = threading.get_ident() + return [object()] + + def get_affected_pages(*args): + worker_threads["affected"] = threading.get_ident() + return SimpleNamespace(regenerate=[]) + + detector.get_changed_files.side_effect = get_changed_files + detector.get_affected_pages.side_effect = get_affected_pages + + def build_graph(): + worker_threads["graph"] = threading.get_ident() + return object() + + result = SimpleNamespace( + file_count=10, + graph_builder=SimpleNamespace(graph=build_graph), + ) + + with ( + patch("subprocess.run", side_effect=head_lookup), + patch("repowise.core.ingestion.ChangeDetector", return_value=detector), + ): + pages = await _incremental_page_regen( + Path(repo_path), + result, + llm_client=object(), + job_config={}, + progress=None, + ) + + assert pages == [] + assert set(worker_threads) == {"head", "diff", "graph", "affected"} + assert all(thread_id != main_thread for thread_id in worker_threads.values()) diff --git a/tests/unit/server/test_scheduler_async.py b/tests/unit/server/test_scheduler_async.py new file mode 100644 index 000000000..d285cdc3c --- /dev/null +++ b/tests/unit/server/test_scheduler_async.py @@ -0,0 +1,44 @@ +"""Async-safety tests for APScheduler background jobs.""" + +from __future__ import annotations + +import asyncio +import json +import time +from types import SimpleNamespace +from unittest.mock import patch + +from repowise.core.persistence import crud +from repowise.core.persistence.database import get_session +from repowise.server.scheduler import setup_scheduler + + +async def test_polling_fallback_does_not_block_event_loop(session_factory, tmp_path) -> None: + """A slow git HEAD lookup should yield control to other async work.""" + repo_path = tmp_path / "repo" + state_path = repo_path / ".repowise" / "state.json" + state_path.parent.mkdir(parents=True) + state_path.write_text(json.dumps({"last_sync_commit": "same-sha"}), encoding="utf-8") + + async with get_session(session_factory) as session: + await crud.upsert_repository( + session, + name="test-repo", + local_path=str(repo_path), + ) + + loop = asyncio.get_running_loop() + event_loop_progressed = asyncio.Event() + + def slow_head_lookup(*args, **kwargs): + loop.call_soon_threadsafe(event_loop_progressed.set) + time.sleep(0.05) + return SimpleNamespace(returncode=0, stdout="same-sha\n") + + scheduler = setup_scheduler(session_factory) + polling_job = next(job for job in scheduler.get_jobs() if job.id == "polling_fallback") + + with patch("subprocess.run", side_effect=slow_head_lookup): + await polling_job.func() + + assert event_loop_progressed.is_set()