From eb30781e8ed7517fd9089aec46ff443b76d2cec8 Mon Sep 17 00:00:00 2001 From: Eldon Marks Date: Tue, 4 Aug 2026 14:16:46 -0400 Subject: [PATCH 1/2] Rebuild the Postgres pool when the event loop changes asyncpg pools and asyncio.Lock both bind to the loop they were created on, but _ensure_pool() memoized the pool and lock for the lifetime of the PostgresDB instance. A host that bootstraps inside one asyncio.run() and then serves from a second loop -- the common CLI-then-ASGI-server startup -- carried a pool bound to the first, now-closed loop into the server loop, and its first query failed with "cannot perform operation: another operation is in progress" / ConnectionDoesNotExistError. Track the loop the pool belongs to and drop both pool and lock when the running loop differs. The stale pool's connections are abandoned rather than closed, since close() would have to await on a loop that no longer runs; Postgres reaps them when the sockets drop. File-backed adapters never hit this, which is why it went unnoticed: it only bites the Server path on postgres. --- jvspatial/db/postgres.py | 37 +++++++++++++++ tests/db/test_postgres_unit.py | 84 ++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/jvspatial/db/postgres.py b/jvspatial/db/postgres.py index 690615b..8f01d70 100644 --- a/jvspatial/db/postgres.py +++ b/jvspatial/db/postgres.py @@ -337,6 +337,10 @@ def __init__( self._pool: Optional["Pool"] = None self._pool_lock = asyncio.Lock() + # The loop ``_pool`` (and ``_pool_lock``) belong to. asyncpg pools + # and asyncio primitives bind to the loop they were created on, so + # a pool carried into a second loop is unusable. + self._pool_loop: Optional[asyncio.AbstractEventLoop] = None # Collections we've already created the table + base indexes for. # Avoids running CREATE TABLE IF NOT EXISTS on the hot path. @@ -350,8 +354,39 @@ def __init__( # ---- pool lifecycle ---------------------------------------------------- + def _discard_pool_from_dead_loop(self) -> None: + """Drop a pool that belongs to a different event loop. + + Hosts that bootstrap inside one ``asyncio.run()`` and then serve + from a second loop (the CLI-then-ASGI-server pattern) would + otherwise reuse a pool bound to the first, now-closed loop, and + every query fails with ``cannot perform operation: another + operation is in progress``. The stale pool's connections are + abandoned rather than closed — ``close()`` would have to await on + the dead loop — and Postgres reaps them when the sockets drop. + """ + if self._pool is None: + return + try: + running = asyncio.get_running_loop() + except RuntimeError: # pragma: no cover - callers are async + return + if self._pool_loop is running: + return + + logger.debug( + "PostgresDB: event loop changed; rebuilding pool (was %r, now %r)", + self._pool_loop, + running, + ) + self._pool = None + # ``asyncio.Lock`` binds to its loop too, so it has to go as well. + self._pool_lock = asyncio.Lock() + self._collections_bootstrapped.clear() + async def _ensure_pool(self) -> "Pool": """Lazily create the asyncpg pool. Idempotent + concurrency-safe.""" + self._discard_pool_from_dead_loop() if self._pool is not None: return self._pool async with self._pool_lock: @@ -377,6 +412,7 @@ async def _ensure_pool(self) -> "Pool": self.pooler_mode, ) self._pool = await asyncpg.create_pool(**create_kwargs) + self._pool_loop = asyncio.get_running_loop() return self._pool async def close(self) -> None: @@ -384,6 +420,7 @@ async def close(self) -> None: if self._pool is not None: await self._pool.close() self._pool = None + self._pool_loop = None self._collections_bootstrapped.clear() # ---- tenant scope (C6) ------------------------------------------------- diff --git a/tests/db/test_postgres_unit.py b/tests/db/test_postgres_unit.py index aeb3099..d9df858 100644 --- a/tests/db/test_postgres_unit.py +++ b/tests/db/test_postgres_unit.py @@ -248,3 +248,87 @@ def test_malformed_vector_left_to_translator_to_reject(self) -> None: filtered, field, _, _, _ = db._pop_vector_clause("doc", q) assert filtered == q # untouched assert field is None + + +class TestPoolLoopAffinity: + """A pool belongs to the loop that created it. + + Hosts that bootstrap in one ``asyncio.run()`` and then serve from a + second loop (the CLI-then-uvicorn pattern) would otherwise reuse a + pool bound to a dead loop, and every query fails with + ``cannot perform operation: another operation is in progress``. + """ + + @staticmethod + def _fake_pool() -> object: + class _Pool: + pass + + return _Pool() + + def test_pool_reused_within_one_loop(self, monkeypatch: pytest.MonkeyPatch) -> None: + db = PostgresDB(dsn="postgresql://nope/none") + created = [] + + async def fake_create_pool(**_kwargs: object) -> object: + pool = self._fake_pool() + created.append(pool) + return pool + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + async def scenario() -> None: + first = await db._ensure_pool() + second = await db._ensure_pool() + assert first is second + + asyncio.run(scenario()) + assert len(created) == 1 + + def test_pool_rebuilt_after_loop_change( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + db = PostgresDB(dsn="postgresql://nope/none") + created = [] + + async def fake_create_pool(**_kwargs: object) -> object: + pool = self._fake_pool() + created.append(pool) + return pool + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + pools: list = [] + # Two separate asyncio.run() calls == two distinct event loops. + asyncio.run(_collect_pool(db, pools)) + asyncio.run(_collect_pool(db, pools)) + + assert len(created) == 2 + assert pools[0] is not pools[1] + + def test_lock_is_not_bound_to_the_dead_loop( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The rebuild must replace ``_pool_lock`` too. + + ``asyncio.Lock`` binds to the loop it is first awaited on, so + carrying the old lock across would trade one cross-loop failure + for another. + """ + db = PostgresDB(dsn="postgresql://nope/none") + + async def fake_create_pool(**_kwargs: object) -> object: + return self._fake_pool() + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + pools: list = [] + asyncio.run(_collect_pool(db, pools)) + first_lock = db._pool_lock + + asyncio.run(_collect_pool(db, pools)) + assert db._pool_lock is not first_lock + + +async def _collect_pool(db: PostgresDB, sink: list) -> None: + sink.append(await db._ensure_pool()) From 8b739068e202dea817eb3564ba9e8540feb75f4e Mon Sep 17 00:00:00 2001 From: Eldon Marks Date: Tue, 4 Aug 2026 14:16:57 -0400 Subject: [PATCH 2/2] Support postgres as a Server database type DatabaseConfigurator.initialize_graph_context() dispatched db_type through a hard-coded json/mongodb/sqlite/dynamodb chain and raised "Unsupported database type: postgres" for anything else. PostgresDB and create_database("postgres", ...) were already complete, and the JVSPATIAL_POSTGRES_* keys were already allowlisted -- but Server() is the only path most deployments use, so a documented backend was unreachable from the API layer and from anything built on it. Add the missing branch, and give ServerConfig.database the settings it needs to describe the connection: postgres_dsn, postgres_min_pool_size, postgres_max_pool_size, postgres_pooler_mode, mapped from env in env_adapter alongside the mongodb and dynamodb keys. Connection settings now reach the driver through the config object rather than only through the driver's own env reads. Unset values are omitted so PostgresDB's defaults still apply. postgresql is accepted as an alias for postgres, matching the factory. --- CHANGELOG.md | 26 ++++++ docs/md/postgres-guide.md | 40 +++++++++ .../api/components/database_configurator.py | 25 ++++++ jvspatial/api/config_groups.py | 17 ++++ jvspatial/env_adapter.py | 8 ++ .../components/test_database_configurator.py | 84 +++++++++++++++++++ tests/test_env_adapter_postgres.py | 53 ++++++++++++ 7 files changed, 253 insertions(+) create mode 100644 tests/test_env_adapter_postgres.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8337d38..0e91351 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Postgres is selectable from `Server`** (`jvspatial/api/components/database_configurator.py`). + `initialize_graph_context()` now builds a `PostgresDB` prime database for + `db_type="postgres"` (alias `"postgresql"`) instead of raising + `ValueError: Unsupported database type: postgres`. The backend and + `create_database("postgres", ...)` already worked; only the `Server` path + was missing, so every API-layer deployment — and anything built on it — was + locked out of a documented backend. + +- **`ServerConfig.database` carries Postgres settings** (`jvspatial/api/config_groups.py`, + `jvspatial/env_adapter.py`) — `postgres_dsn`, `postgres_min_pool_size`, + `postgres_max_pool_size`, `postgres_pooler_mode`, populated from the + already-allowlisted `JVSPATIAL_POSTGRES_*` env keys. Connection settings now + flow through the config object like every other backend's rather than being + readable only by the driver. Unset values still defer to `PostgresDB`'s own + defaults. Coverage: `tests/test_env_adapter_postgres.py`, + `tests/api/components/test_database_configurator.py`. + - **`resolve_sort_value(record, field)`** (`jvspatial/db/database.py`) — the dotted-path resolution `finalize_find_results` uses, exported so adapters and cursor logic resolve a sort field the same way. Added to the module's @@ -24,6 +41,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`PostgresDB` pool is event-loop aware** (`jvspatial/db/postgres.py`). The + `asyncpg` pool and its lock bind to the loop that created them, but + `_ensure_pool()` memoized both for the lifetime of the instance. A host that + bootstrapped in one `asyncio.run()` and then served from a second loop hit + `cannot perform operation: another operation is in progress` / + `ConnectionDoesNotExistError` on its first query. The pool and lock are now + rebuilt when the running loop changes. Coverage: + `tests/db/test_postgres_unit.py::TestPoolLoopAffinity`. + - **Partial-index repair log is INFO, not WARNING** (`jvspatial/db/sqlite.py`). Dropping a non-partial index so it can be recreated with `WHERE` is expected one-shot migration noise; log at info. Also satisfy ruff SIM110 in diff --git a/docs/md/postgres-guide.md b/docs/md/postgres-guide.md index 6bbf593..06b7fb5 100644 --- a/docs/md/postgres-guide.md +++ b/docs/md/postgres-guide.md @@ -77,6 +77,37 @@ alice = await User.create(name="Alice", email="alice@example.com") loaded = await User.get(alice.id) ``` +### Using Postgres with `Server` + +`Server` selects the prime database from `db_type`, so Postgres is available +to the API layer the same way JSON or MongoDB is: + +```python +from jvspatial.api.server import Server + +server = Server( + db_type="postgres", + postgres_dsn="postgresql://user:pw@localhost:5432/mydb", +) +``` + +`postgresql` is accepted as an alias for `postgres`. Every connection setting +also resolves from the environment through `ServerConfig`: + +```bash +JVSPATIAL_DB_TYPE=postgres +JVSPATIAL_POSTGRES_DSN=postgresql://user:pw@localhost:5432/mydb +``` + +Settings you leave unset fall through to `PostgresDB`'s own defaults, so the +DSN is the only value most deployments need. `JVSPATIAL_DB_PATH` does not +apply — it is a file-backend setting. + +Note that the logging database is a separate store and has no Postgres +backend; if `JVSPATIAL_LOG_DB_TYPE` is unset it inherits `JVSPATIAL_DB_TYPE` +and falls back to a JSON file log. Set it explicitly when the graph runs on +Postgres. + ## Schema Each collection becomes one table. The shape: @@ -152,6 +183,15 @@ JVSPATIAL_POSTGRES_MIN_POOL_SIZE=5 JVSPATIAL_POSTGRES_MAX_POOL_SIZE=25 ``` +### Event loops + +The pool belongs to the event loop that created it. If a host bootstraps +inside one `asyncio.run()` and then serves from a second loop — the common +CLI-then-ASGI-server startup — `PostgresDB` notices the loop change and +rebuilds the pool on first use in the new loop. The abandoned pool's +connections are dropped rather than closed, since closing them would mean +awaiting on a loop that no longer runs. + ### Pooler compatibility (PgBouncer / RDS Proxy) Transaction-mode poolers (PgBouncer transaction-pooling, AWS RDS Proxy) reuse diff --git a/jvspatial/api/components/database_configurator.py b/jvspatial/api/components/database_configurator.py index a486c64..1bf88cb 100644 --- a/jvspatial/api/components/database_configurator.py +++ b/jvspatial/api/components/database_configurator.py @@ -79,6 +79,25 @@ def _resolve_mongodb_connection(self) -> Tuple[str, str]: db_name = (db.db_database_name or "").strip() or "jvdb" return uri, db_name + def _resolve_postgres_kwargs(self) -> Dict[str, Any]: + """Resolve PostgresDB connection kwargs from server configuration. + + Only settings the operator actually supplied are returned, so + anything left unset falls through to ``PostgresDB``'s own env + defaults rather than being pinned to a value here. + """ + db = self.config.database + kwargs: Dict[str, Any] = {} + if db.postgres_dsn: + kwargs["dsn"] = db.postgres_dsn + if db.postgres_min_pool_size is not None: + kwargs["min_size"] = db.postgres_min_pool_size + if db.postgres_max_pool_size is not None: + kwargs["max_size"] = db.postgres_max_pool_size + if db.postgres_pooler_mode: + kwargs["pooler_mode"] = db.postgres_pooler_mode + return kwargs + def _resolve_observability_kwargs(self) -> Dict[str, Any]: """Resolve optional DB observability wrapper kwargs from env.""" raw_enabled = str(os.environ.get("JVSPATIAL_OBSERVABILITY_ENABLED", "")).lower() @@ -173,6 +192,12 @@ def initialize_graph_context(self) -> Optional[GraphContext]: aws_secret_access_key=self.config.database.dynamodb_secret_access_key, **observability_kwargs, ) + elif db_type in ("postgres", "postgresql"): + prime_db = create_database( + db_type="postgres", + **self._resolve_postgres_kwargs(), + **observability_kwargs, + ) else: raise ValueError(f"Unsupported database type: {db_type}") diff --git a/jvspatial/api/config_groups.py b/jvspatial/api/config_groups.py index 7aef8ab..3105cf7 100644 --- a/jvspatial/api/config_groups.py +++ b/jvspatial/api/config_groups.py @@ -45,6 +45,23 @@ class DatabaseConfig(BaseModel): default=None, validation_alias="AWS_SECRET_ACCESS_KEY" ) + # PostgreSQL Configuration (only used if db_type is "postgres"/"postgresql"). + # Unset values fall through to PostgresDB's own env defaults. + postgres_dsn: Optional[str] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_DSN" + ) + postgres_min_pool_size: Optional[int] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_MIN_POOL_SIZE" + ) + postgres_max_pool_size: Optional[int] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_MAX_POOL_SIZE" + ) + postgres_pooler_mode: Optional[str] = Field( + default=None, + validation_alias="JVSPATIAL_POSTGRES_POOLER_MODE", + description="'session' (default) or 'transaction' for PgBouncer / RDS Proxy.", + ) + class SecurityConfig(BaseModel): """Security configuration group.""" diff --git a/jvspatial/env_adapter.py b/jvspatial/env_adapter.py index 1593b5d..c65523d 100644 --- a/jvspatial/env_adapter.py +++ b/jvspatial/env_adapter.py @@ -124,9 +124,17 @@ def server_config_overrides_from_env() -> Dict[str, Any]: ("JVSPATIAL_DYNAMODB_TABLE_NAME", "dynamodb_table_name"), ("JVSPATIAL_DYNAMODB_REGION", "dynamodb_region"), ("JVSPATIAL_DYNAMODB_ENDPOINT_URL", "dynamodb_endpoint_url"), + ("JVSPATIAL_POSTGRES_DSN", "postgres_dsn"), + ("JVSPATIAL_POSTGRES_POOLER_MODE", "postgres_pooler_mode"), ): if (t := _opt_str(ek)) is not None: db[dk] = t + for ek, dk in ( + ("JVSPATIAL_POSTGRES_MIN_POOL_SIZE", "postgres_min_pool_size"), + ("JVSPATIAL_POSTGRES_MAX_POOL_SIZE", "postgres_max_pool_size"), + ): + if (n := _opt_int(ek)) is not None: + db[dk] = n if db: o["database"] = db diff --git a/tests/api/components/test_database_configurator.py b/tests/api/components/test_database_configurator.py index 05b72c7..9b4c5a0 100644 --- a/tests/api/components/test_database_configurator.py +++ b/tests/api/components/test_database_configurator.py @@ -249,6 +249,90 @@ async def test_initialize_graph_context_dynamodb(self): slow_query_ms=100.0, ) + @pytest.mark.asyncio + async def test_initialize_graph_context_postgres(self): + """Test GraphContext initialization with PostgreSQL.""" + config = ServerConfig() + config.database.db_type = "postgres" + config.database.postgres_dsn = "postgresql://u:p@localhost:5432/testdb" + config.database.postgres_min_pool_size = 3 + config.database.postgres_max_pool_size = 7 + config.database.postgres_pooler_mode = "transaction" + configurator = DatabaseConfigurator(config) + + with patch( + "jvspatial.api.components.database_configurator.create_database" + ) as mock_create: + mock_db = MagicMock() + mock_create.return_value = mock_db + + with patch( + "jvspatial.api.components.database_configurator.get_database_manager" + ) as mock_get_manager: + mock_manager = MagicMock() + mock_manager.get_current_database.return_value = mock_db + mock_get_manager.side_effect = RuntimeError() + + with patch( + "jvspatial.api.components.database_configurator.set_database_manager" + ): + with patch( + "jvspatial.api.components.database_configurator.GraphContext" + ) as mock_context: + mock_ctx_instance = MagicMock() + mock_context.return_value = mock_ctx_instance + + with patch( + "jvspatial.api.components.database_configurator.set_default_context" + ): + result = configurator.initialize_graph_context() + + assert result == mock_ctx_instance + mock_create.assert_called_once_with( + db_type="postgres", + dsn="postgresql://u:p@localhost:5432/testdb", + min_size=3, + max_size=7, + pooler_mode="transaction", + observe=False, + slow_query_ms=100.0, + ) + + @pytest.mark.asyncio + async def test_initialize_graph_context_postgresql_alias(self): + """``postgresql`` is accepted as an alias for ``postgres``.""" + config = ServerConfig() + config.database.db_type = "postgresql" + configurator = DatabaseConfigurator(config) + + with patch( + "jvspatial.api.components.database_configurator.create_database" + ) as mock_create: + mock_create.return_value = MagicMock() + with patch( + "jvspatial.api.components.database_configurator.get_database_manager", + side_effect=RuntimeError(), + ): + with patch( + "jvspatial.api.components.database_configurator.set_database_manager" + ): + with patch( + "jvspatial.api.components.database_configurator.GraphContext", + return_value=MagicMock(), + ): + with patch( + "jvspatial.api.components.database_configurator.set_default_context" + ): + configurator.initialize_graph_context() + + # Unset connection settings are omitted so PostgresDB's own env + # defaults still apply. + mock_create.assert_called_once_with( + db_type="postgres", + observe=False, + slow_query_ms=100.0, + ) + @pytest.mark.asyncio async def test_initialize_graph_context_no_db_type(self, configurator): """Test that None is returned when db_type is not set.""" diff --git a/tests/test_env_adapter_postgres.py b/tests/test_env_adapter_postgres.py new file mode 100644 index 0000000..faae296 --- /dev/null +++ b/tests/test_env_adapter_postgres.py @@ -0,0 +1,53 @@ +"""``JVSPATIAL_POSTGRES_*`` env keys map onto ServerConfig.database.""" + +from __future__ import annotations + +import pytest + +from jvspatial.env_adapter import server_config_overrides_from_env + +_POSTGRES_ENV_KEYS = ( + "JVSPATIAL_POSTGRES_DSN", + "JVSPATIAL_POSTGRES_MIN_POOL_SIZE", + "JVSPATIAL_POSTGRES_MAX_POOL_SIZE", + "JVSPATIAL_POSTGRES_POOLER_MODE", +) + + +@pytest.fixture(autouse=True) +def _clear_postgres_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Host ``.env`` values must not leak into these assertions.""" + for key in _POSTGRES_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + + +def test_postgres_env_maps_into_database_group( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "postgres") + monkeypatch.setenv("JVSPATIAL_POSTGRES_DSN", "postgresql://u:p@host:5432/db") + monkeypatch.setenv("JVSPATIAL_POSTGRES_MIN_POOL_SIZE", "1") + monkeypatch.setenv("JVSPATIAL_POSTGRES_MAX_POOL_SIZE", "4") + monkeypatch.setenv("JVSPATIAL_POSTGRES_POOLER_MODE", "transaction") + + db = server_config_overrides_from_env()["database"] + + assert db["db_type"] == "postgres" + assert db["postgres_dsn"] == "postgresql://u:p@host:5432/db" + assert db["postgres_min_pool_size"] == 1 + assert db["postgres_max_pool_size"] == 4 + assert db["postgres_pooler_mode"] == "transaction" + + +def test_postgres_keys_absent_when_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "json") + + db = server_config_overrides_from_env().get("database", {}) + + for key in ( + "postgres_dsn", + "postgres_min_pool_size", + "postgres_max_pool_size", + "postgres_pooler_mode", + ): + assert key not in db