diff --git a/packages/core/src/repowise/core/persistence/database.py b/packages/core/src/repowise/core/persistence/database.py index ba675492c..cdf3c46cd 100644 --- a/packages/core/src/repowise/core/persistence/database.py +++ b/packages/core/src/repowise/core/persistence/database.py @@ -237,6 +237,7 @@ def create_engine( # Pass use_static_pool=True explicitly when creating in-memory test engines. use_static_pool: bool = False, busy_timeout_ms: int | None = None, + short_lived: bool = True, ) -> AsyncEngine: """Create an AsyncEngine for the given database URL. @@ -249,6 +250,25 @@ def create_engine( small value for best-effort secondary writers that must never stall the primary writer (issue #326). Ignored for non-SQLite backends. + short_lived: Whether this engine is created, used, and disposed + within a single call (the pattern almost every caller + follows: one CLI command, one workspace update, one + background task). Defaults to True, which uses + NullPool for PostgreSQL — one connection per checkout, + closed on dispose, so a short-lived engine can never + hold more than a single Postgres server slot, and an + engine that outlives its creating event loop can never + hand back a dead pooled connection to a later one + (issue #2062's failure class). Pass False only for an + engine stored for a process's lifetime and reused + across many requests — currently just the FastAPI app + and the MCP server — where SQLAlchemy's pooled + AsyncAdaptedQueuePool is the correct choice and + NullPool would open a fresh connection per request. + Ignored for SQLite, which already always uses + NullPool (or StaticPool for :memory:) regardless of + this flag — SQLite has no equivalent long-lived-pool + need since ``aiosqlite`` connections are cheap. """ db_url = get_db_url(url) is_sqlite = db_url.startswith("sqlite") @@ -265,8 +285,21 @@ def create_engine( else: kwargs["poolclass"] = NullPool else: - # PostgreSQL — asyncpg handles its own connection pool + # PostgreSQL. SQLAlchemy pools these connections with + # AsyncAdaptedQueuePool by default — asyncpg does NOT provide its own + # pool here (that only happens if something calls asyncpg.create_pool, + # which nothing in this codebase does). Every create_engine() call in + # this codebase except the long-lived server/MCP engines is + # short-lived (create, use, dispose within one async function), so + # there's no reuse to gain from pooling and every pooled-but-idle + # connection is a Postgres server slot held for no benefit — or, + # worse, one that survives past a closed event loop and gets handed + # to a later, unrelated caller (#2062's failure class). NullPool caps + # a short-lived engine's footprint at exactly one connection instead + # of up to 15 (pool_size=5 + max_overflow=10) sitting idle. kwargs["pool_pre_ping"] = True + if short_lived: + kwargs["poolclass"] = NullPool engine = create_async_engine(db_url, **kwargs) if is_sqlite: diff --git a/packages/core/src/repowise/core/workspace/update.py b/packages/core/src/repowise/core/workspace/update.py index 6714c1220..fe286b068 100644 --- a/packages/core/src/repowise/core/workspace/update.py +++ b/packages/core/src/repowise/core/workspace/update.py @@ -51,26 +51,73 @@ _log = logging.getLogger("repowise.workspace.update") -def _merged_repo_excludes( +async def _merged_repo_excludes( repo_path: Path, extra_exclude_patterns: list[str] | None = None, ) -> list[str]: + """Merge config.yaml excludes with the persisted repo settings' excludes. + + Local SQLite indexes store ``settings_json`` in the repo-local + ``wiki.db``, read directly since this call is on the hot per-update path + and a raw sqlite3 read is cheaper than opening an engine for one row. + When a shared database is configured, that DB is the source of truth + instead — reading only ``config.yaml`` there would silently drop every + pattern the user added through the shared repository's settings, so + every workspace update would run with a narrower exclude set than the + repo was actually indexed with. + """ from ..repo_config import load_repo_config patterns: list[str] = list(load_repo_config(repo_path).get("exclude_patterns") or []) - db_path = repo_path / ".repowise" / "wiki.db" - if db_path.is_file(): + + from ..persistence.database import get_configured_db_url + + configured_url = get_configured_db_url() + settings_json: str | None = None + + if configured_url is None: + db_path = repo_path / ".repowise" / "wiki.db" + if db_path.is_file(): + try: + with sqlite3.connect(str(db_path)) as conn: + row = conn.execute("SELECT settings_json FROM repositories LIMIT 1").fetchone() + if row and row[0]: + settings_json = row[0] + except Exception: + pass + else: + from ..persistence import ( + create_engine, + create_session_factory, + get_session, + init_db, + ) + from ..persistence.crud import get_repository_by_path + + try: + engine = create_engine(configured_url) + try: + await init_db(engine) + sf = create_session_factory(engine) + async with get_session(sf) as session: + repo = await get_repository_by_path(session, str(repo_path)) + if repo is not None and repo.settings_json: + settings_json = repo.settings_json + finally: + await engine.dispose() + except Exception: + pass + + if settings_json: try: - with sqlite3.connect(str(db_path)) as conn: - row = conn.execute("SELECT settings_json FROM repositories LIMIT 1").fetchone() - if row and row[0]: - settings = _json.loads(row[0]) - if isinstance(settings, dict): - for value in settings.get("exclude_patterns") or []: - if isinstance(value, str) and value not in patterns: - patterns.append(value) + settings = _json.loads(settings_json) + if isinstance(settings, dict): + for value in settings.get("exclude_patterns") or []: + if isinstance(value, str) and value not in patterns: + patterns.append(value) except Exception: pass + for pattern in extra_exclude_patterns or []: if pattern not in patterns: patterns.append(pattern) @@ -305,16 +352,26 @@ async def reconcile_repo_head_commit(repo_path: Path, head: str | None) -> None: sync-check — a routine ``repowise update`` that finds nothing to do still counts as "verified current now". Creates the row when it is missing from an existing ``wiki.db`` (self-heals a corrupt/blank store — the policy the - CLI's ``stamp_head_commit``, now a thin wrapper over this, always had); + CLI's ``stamp_head_commit``, now a thin wrapper over this, always had). + + When a shared database is configured (``get_configured_db_url()``), that + DB is the source of truth and this always runs — a repo-local ``wiki.db`` + is not expected to exist in that mode. Otherwise (local SQLite), this is still a no-op when ``wiki.db`` itself is absent, so a stamp can never - conjure an empty database. + conjure an empty local database. This is the single head-commit stamper for both update paths — the CLI fast paths and the workspace updater used to run two implementations with different creation semantics. """ - if not head or not (repo_path / ".repowise" / "wiki.db").is_file(): + if not head: + return + from ..persistence.database import get_configured_db_url + + configured_url = get_configured_db_url() + if configured_url is None and not (repo_path / ".repowise" / "wiki.db").is_file(): return + from ..persistence import ( create_engine, create_session_factory, @@ -325,7 +382,7 @@ async def reconcile_repo_head_commit(repo_path: Path, head: str | None) -> None: from ..persistence.crud import get_repository_by_path from ..persistence.database import resolve_db_url - url = resolve_db_url(repo_path) + url = configured_url or resolve_db_url(repo_path) engine = create_engine(url) try: await init_db(engine) @@ -407,16 +464,14 @@ async def _incremental_repo_update( # New commits but nothing the index cares about changed (merge/empty # commits, or every change excluded). Report success so the caller # bumps ``last_sync_commit`` instead of re-diffing forever. - return RepoUpdateResult( - alias=alias, updated=True, working_tree_paths=working_tree_paths - ) + return RepoUpdateResult(alias=alias, updated=True, working_tree_paths=working_tree_paths) # Per-repo config, like the single-repo update path. The workspace-level # ``exclude_patterns`` (when provided) apply on top. from ..repo_config import load_repo_config cfg = load_repo_config(repo_path) - merged_excludes = _merged_repo_excludes(repo_path, exclude_patterns) + merged_excludes = await _merged_repo_excludes(repo_path, exclude_patterns) # Decay-only rows for idle files the anchor advance recovered (#728); # persisted alongside the changed rows, kept out of git_meta_map so partial @@ -542,6 +597,36 @@ async def _incremental_repo_update( ) +async def _has_persisted_repo_index(repo_path: Path) -> bool: + """Return whether this repo has persisted index data. + + Local SQLite indexes are identified by their repo-local wiki.db. + When a shared database is configured, the repository row is the source + of truth instead; a repo-local wiki.db is not expected to exist. + """ + from ..persistence.database import get_configured_db_url + + if get_configured_db_url() is None: + return (repo_path / ".repowise" / "wiki.db").is_file() + + from ..persistence import ( + create_engine, + create_session_factory, + get_session, + init_db, + ) + from ..persistence.crud import get_repository_by_path + + engine = create_engine(get_configured_db_url()) + try: + await init_db(engine) + sf = create_session_factory(engine) + async with get_session(sf) as session: + return await get_repository_by_path(session, str(repo_path)) is not None + finally: + await engine.dispose() + + async def update_single_repo_index( repo_path: Path, *, @@ -565,7 +650,7 @@ async def update_single_repo_index( alias = repo_path.name state = read_repo_state(repo_path) base_ref = state.get("last_sync_commit") - merged_excludes = _merged_repo_excludes(repo_path, exclude_patterns) + merged_excludes = await _merged_repo_excludes(repo_path, exclude_patterns) # Config drift check, mirroring the single-repo update path: a changed # config.yaml / health-rules.json invalidates persisted health scores and @@ -577,7 +662,9 @@ async def update_single_repo_index( # surprise full re-index. stored_fp = state.get("config_fingerprint") config_changed = stored_fp is not None and stored_fp != config_fingerprint(repo_path) - if config_changed and (repo_path / ".repowise" / "wiki.db").is_file(): + + has_persisted_index = await _has_persisted_repo_index(repo_path) + if config_changed and has_persisted_index: _log.info( "workspace_update: %s config fingerprint drifted — full re-index " "so health scores reflect the new config", @@ -587,7 +674,7 @@ async def update_single_repo_index( if ( not config_changed and base_ref - and (repo_path / ".repowise" / "wiki.db").is_file() + and has_persisted_index and commit_exists(repo_path, str(base_ref)) ): try: @@ -798,9 +885,7 @@ async def update_workspace( if new_head: with suppress(OSError): (path / ".repowise").mkdir(parents=True, exist_ok=True) - (path / ".repowise" / ".update.pending").write_text( - new_head, encoding="utf-8" - ) + (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8") return [ RepoUpdateResult( alias=alias, @@ -848,7 +933,9 @@ async def _update_one( ) # Record pending so the running update can roll forward. with suppress(OSError): - (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8") + (path / ".repowise" / ".update.pending").write_text( + new_head, encoding="utf-8" + ) return RepoUpdateResult( alias=alias, updated=False, @@ -1019,9 +1106,7 @@ async def run_cross_repo_hooks( else WorkspaceIndex({}) ) try: - await _run_phases( - ws_config, workspace_root, changed_repos, timings, workspace_index - ) + await _run_phases(ws_config, workspace_root, changed_repos, timings, workspace_index) finally: await workspace_index.close() @@ -1087,7 +1172,9 @@ async def _timed(phase: str, coro: Awaitable[Any]) -> Any: timings.on_phase_done(phase) overlay_result, store_result = await asyncio.gather( - _timed("cross_repo_analysis", run_cross_repo_analysis(ws_config, workspace_root, changed_repos)), + _timed( + "cross_repo_analysis", run_cross_repo_analysis(ws_config, workspace_root, changed_repos) + ), _timed( "contract_extraction", run_contract_extraction( diff --git a/packages/server/src/repowise/server/app.py b/packages/server/src/repowise/server/app.py index 9449f8a9b..e7a785d30 100644 --- a/packages/server/src/repowise/server/app.py +++ b/packages/server/src/repowise/server/app.py @@ -184,7 +184,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: pass # Fall back to default - engine = create_engine(db_url) + engine = create_engine(db_url, short_lived=False) await init_db(engine) session_factory = create_session_factory(engine) @@ -327,6 +327,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: app.state.workspace_fts[repo_id] = fts continue + # NullPool is irrelevant here since SQLite ignores create_engine()'s + # short_lived flag — but this engine is stored in app.state for the + # server's lifetime (workspace_engines/workspace_sessions), matching the + # long-lived pattern in the two engines above. If this URL is ever made + # configurable (e.g. pointed at a shared REPOWISE_DB_URL for workspace + # members), it must pass short_lived=False too. repo_engine = create_engine(f"sqlite+aiosqlite:///{db_url_posix}") await init_db(repo_engine) repo_sf = create_session_factory(repo_engine) diff --git a/packages/server/src/repowise/server/mcp_server/_server.py b/packages/server/src/repowise/server/mcp_server/_server.py index e68f4ff72..6e61a90cb 100644 --- a/packages/server/src/repowise/server/mcp_server/_server.py +++ b/packages/server/src/repowise/server/mcp_server/_server.py @@ -527,7 +527,7 @@ async def _lifespan(server: FastMCP): db_url = resolve_db_url(_state._repo_path) _log.info("repowise MCP: initialising database…") - engine = create_engine(db_url) + engine = create_engine(db_url, short_lived=False) await init_db(engine) _state._session_factory = async_sessionmaker( diff --git a/tests/unit/server/mcp/test_config.py b/tests/unit/server/mcp/test_config.py index 8dadcd6dd..969f6c510 100644 --- a/tests/unit/server/mcp/test_config.py +++ b/tests/unit/server/mcp/test_config.py @@ -62,8 +62,9 @@ async def fake_init_db(engine) -> None: async def fake_load_vector_stores(repo_path: str | None) -> None: return None - def fake_create_engine(url: str) -> DummyEngine: + def fake_create_engine(url: str, **kwargs) -> DummyEngine: captured["url"] = url + captured["short_lived"] = kwargs.get("short_lived") return DummyEngine() monkeypatch.setenv("REPOWISE_DB_URL", "sqlite+aiosqlite:///tmp/from-cli.db") @@ -88,6 +89,7 @@ def fake_create_engine(url: str) -> DummyEngine: try: async with mcp_server._lifespan(mcp_server.mcp): assert captured["url"] == "sqlite+aiosqlite:///tmp/from-cli.db" + assert captured["short_lived"] is False finally: _state._repo_path = original_repo_path _state._vector_store = original_vector_store diff --git a/tests/unit/workspace/test_incremental_update.py b/tests/unit/workspace/test_incremental_update.py index 59de685c3..ce7d3baa8 100644 --- a/tests/unit/workspace/test_incremental_update.py +++ b/tests/unit/workspace/test_incremental_update.py @@ -372,3 +372,129 @@ def test_read_repo_state_missing_or_malformed(tmp_path): (tmp_path / ".repowise").mkdir() (tmp_path / ".repowise" / "state.json").write_text("not json{") assert read_repo_state(tmp_path) == {} + + +def test_shared_db_indexed_repo_takes_incremental_path(tmp_path, forbid_full_pipeline, monkeypatch): + """A repo indexed in the configured shared DB must update incrementally + even when no repo-local .repowise/wiki.db exists.""" + repo = _make_git_repo(tmp_path) + base = get_head_commit(repo) + + # Use a file-backed SQLite database as a stand-in for the configured + # external/shared database. The routing decision must depend on the + # configured DB, not on the presence of repo-local wiki.db. + shared_db = tmp_path / "shared.db" + monkeypatch.setenv( + "REPOWISE_DB_URL", + f"sqlite+aiosqlite:///{shared_db}", + ) + + from repowise.core.persistence import ( + create_engine, + create_session_factory, + get_session, + init_db, + upsert_repository, + ) + + async def _seed() -> None: + engine = create_engine(f"sqlite+aiosqlite:///{shared_db}") + try: + await init_db(engine) + sf = create_session_factory(engine) + async with get_session(sf) as session: + await upsert_repository( + session, + name=repo.name, + local_path=str(repo), + head_commit=base, + ) + finally: + await engine.dispose() + + asyncio.run(_seed()) + + # Persist the incremental anchor, but deliberately do NOT create + # /.repowise/wiki.db. + state_dir = repo / ".repowise" + state_dir.mkdir(parents=True, exist_ok=True) + (state_dir / "state.json").write_text( + json.dumps({"last_sync_commit": base}), + encoding="utf-8", + ) + assert not (state_dir / "wiki.db").exists() + + _add_commit(repo, "b.py") + + result = asyncio.run(update_single_repo_index(repo)) + + assert result.error is None + assert result.updated is True + assert result.file_count >= 2 + assert result.symbol_count == 2 + + +def test_shared_db_config_drift_does_not_crash(tmp_path, stub_full_pipeline, monkeypatch): + """Config-fingerprint drift on a shared-DB-indexed repo (no local + wiki.db) must still fall back to the full pipeline cleanly. + + Regression test: an earlier version of the shared-DB gate computed + ``has_persisted_index`` *after* the config-drift branch already read it, + which raised UnboundLocalError the moment config_changed was True for a + repo with no local wiki.db (i.e. exactly the shared-DB case). The + existing test_config_drift_runs_full_reindex only exercises this branch + for the local-SQLite case (wiki.db present via _mark_indexed), so it + never caught the ordering bug. + """ + repo = _make_git_repo(tmp_path) + base = get_head_commit(repo) + + # Use a file-backed SQLite database as a stand-in for the configured + # external/shared database, same as test_shared_db_indexed_repo_takes_incremental_path. + shared_db = tmp_path / "shared.db" + monkeypatch.setenv( + "REPOWISE_DB_URL", + f"sqlite+aiosqlite:///{shared_db}", + ) + + from repowise.core.persistence import ( + create_engine, + create_session_factory, + get_session, + init_db, + upsert_repository, + ) + + async def _seed() -> None: + engine = create_engine(f"sqlite+aiosqlite:///{shared_db}") + try: + await init_db(engine) + sf = create_session_factory(engine) + async with get_session(sf) as session: + await upsert_repository( + session, + name=repo.name, + local_path=str(repo), + head_commit=base, + ) + finally: + await engine.dispose() + + asyncio.run(_seed()) + + # A stored config fingerprint that will not match the freshly computed + # one — this drives config_changed=True. Deliberately do NOT create + # /.repowise/wiki.db: the shared DB is the source of truth here. + state_dir = repo / ".repowise" + state_dir.mkdir(parents=True, exist_ok=True) + (state_dir / "state.json").write_text( + json.dumps({"last_sync_commit": base, "config_fingerprint": "0" * 64}), + encoding="utf-8", + ) + assert not (state_dir / "wiki.db").exists() + + _add_commit(repo, "b.py") + + result = asyncio.run(update_single_repo_index(repo)) + + _assert_full_pipeline_fallback(result, stub_full_pipeline)