diff --git a/.env.example b/.env.example index 922ffb9..4ea1e40 100644 --- a/.env.example +++ b/.env.example @@ -54,6 +54,22 @@ SKIP_DB_LOCK=false # Loopback, link-local and cloud metadata endpoints stay blocked either way. ALLOW_PRIVATE_DB_HOSTS=false +# ── Database connection pool (advanced) ── +# All optional; the defaults shown are what the app uses when these are unset. +# DB_POOL_PRE_PING adds a liveness round trip to the database before handing +# out every pooled connection. It is off by default because that cost lands on +# every request and pool recycling (DB_POOL_RECYCLE below) already retires idle +# connections. Turn it on if you see "connection was closed" errors from a proxy +# that drops idle links, or if you transaction-pool through pgbouncer with a +# server_idle_timeout under DB_POOL_RECYCLE — recycling alone then leaves a gap +# in which a connection can die while idle. +DB_POOL_PRE_PING=false +# Seconds before a pooled connection is retired and reopened. Keep below the +# idle timeout of your server or connection pooler. +DB_POOL_RECYCLE=300 +DB_POOL_SIZE=5 +DB_MAX_OVERFLOW=10 + # ── Maintenance & Storage ── ACTIVITY_LOG_RETENTION_DAYS=30 ACTIVITY_CLEANUP_ENABLED=true diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..70507c3 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,11 @@ +# Shell scripts must keep LF endings even on Windows checkouts. Docker copies +# them into Linux images verbatim, and a CRLF shebang makes the kernel look for +# an interpreter named "/bin/sh\r", which fails with a confusing +# "no such file or directory" on a file that plainly exists. +*.sh text eol=lf +docker-entrypoint.sh text eol=lf + +# Same reasoning for container and tooling files read by Linux. +DOCKERFILE* text eol=lf +Dockerfile* text eol=lf +*.conf text eol=lf diff --git a/backend/DOCKERFILE b/backend/DOCKERFILE index 8cd5508..02c743b 100644 --- a/backend/DOCKERFILE +++ b/backend/DOCKERFILE @@ -48,5 +48,18 @@ HEALTHCHECK --interval=30s --timeout=5s --retries=3 \ # the schema to head before the app serves its first request. ENTRYPOINT ["/app/backend/docker-entrypoint.sh"] +# Bind every address family. "0.0.0.0" is IPv4-only, but the compose network +# runs with enable_ipv6, so DNS for `backend` answers with both an A and an +# AAAA record. nginx uses both, and the AAAA half of its requests hit +# ECONNREFUSED because nothing listens there -- "upstream server temporarily +# disabled" in the error log, and a wasted connection attempt on the way to +# the IPv4 fallback. +# +# The empty host is deliberate and "::" is *not* a substitute: asyncio sets +# IPV6_V6ONLY on any AF_INET6 server socket, so binding "::" listens on IPv6 +# **only** and refuses IPv4 -- trading one broken family for the other. An +# empty host makes asyncio open one socket per family, which is real +# dual-stack. Verified in-container: 0.0.0.0 -> IPv4 only, :: -> IPv6 only, +# "" -> both. CMD ["python", "-m", "uvicorn", "backend.app.server:create_app", \ - "--factory", "--host", "0.0.0.0", "--port", "4321", "--log-level", "warning"] + "--factory", "--host", "", "--port", "4321", "--log-level", "warning"] diff --git a/backend/app/controllers/auth.py b/backend/app/controllers/auth.py index 4b168a8..5adcb4c 100644 --- a/backend/app/controllers/auth.py +++ b/backend/app/controllers/auth.py @@ -38,6 +38,31 @@ _JWKS_CLIENT: PyJWKClient | None = None +# ── Password hashing ──────────────────────────────────────────────────── +# bcrypt is deliberately expensive -- roughly 750ms per call at the default +# cost factor on our container. That cost is the point, but it is CPU-bound +# and releases nothing, so calling it straight from an async handler parks the +# event loop for its whole duration: every other request being served by that +# worker stalls too, not just the one doing the hashing. Push it to the +# threadpool so only the caller waits. + + +async def hash_password(password: str) -> str: + """Hash a password with a fresh salt, off the event loop.""" + return await run_in_threadpool( + lambda: bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode( + "utf-8" + ) + ) + + +async def verify_password(password: str, hashed: str) -> bool: + """Check a password against its bcrypt hash, off the event loop.""" + return await run_in_threadpool( + lambda: bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8")) + ) + + def _get_jwks_client(): global _JWKS_CLIENT if _JWKS_CLIENT is None: @@ -99,8 +124,7 @@ async def login(email: EmailStr, password: str): # Accounts without a password hash (Google/Supabase-only) cannot log in via password. if not user_details.get("hashed_pass"): raise HTTPException(status_code=401, detail="Invalid credentials") - user_bytes = password.encode() - if bcrypt.checkpw(user_bytes, user_details["hashed_pass"].encode("utf-8")): + if await verify_password(password, user_details["hashed_pass"]): return create_jwt(str(user_details["_id"]), user_details["role"]) raise HTTPException(status_code=401, detail="Invalid credentials") @@ -150,10 +174,7 @@ async def signup(user: UserSignup): raise HTTPException( status_code=400, detail="An account with this email already exists" ) - encoded_pass = user.password.encode("utf-8") - salt = bcrypt.gensalt() - hashed_pw = bcrypt.hashpw(encoded_pass, salt) - hashed_pw = hashed_pw.decode("utf-8") + hashed_pw = await hash_password(user.password) user_in_db = UserInDB(email=user.email, hashed_pass=hashed_pw, role=Role.user) try: uid = await create_user(user_in_db) @@ -289,12 +310,8 @@ async def change_password(user_id, old_password, new_password): status_code=400, detail="Social login accounts cannot change password here", ) - old_pass_enc = old_password.encode("utf-8") - new_pass_enc = new_password.encode("utf-8") - if bcrypt.checkpw(old_pass_enc, user_details["hashed_pass"].encode("utf-8")): - salt = bcrypt.gensalt() - new_hashed_pw = bcrypt.hashpw(new_pass_enc, salt) - user_details["hashed_pass"] = new_hashed_pw.decode("utf-8") + if await verify_password(old_password, user_details["hashed_pass"]): + user_details["hashed_pass"] = await hash_password(new_password) await update_user( user_id, hashed_pass=user_details["hashed_pass"], @@ -516,7 +533,6 @@ async def reset_password(token: str, new_password: str): .values(consumed=True) ) - salt = bcrypt.gensalt() - new_hashed = bcrypt.hashpw(new_password.encode("utf-8"), salt).decode("utf-8") + new_hashed = await hash_password(new_password) await update_user(user_id, hashed_pass=new_hashed) return True diff --git a/backend/app/controllers/system.py b/backend/app/controllers/system.py index d7b43ba..42334e2 100644 --- a/backend/app/controllers/system.py +++ b/backend/app/controllers/system.py @@ -1,6 +1,8 @@ import os import logging +import time +import httpx from fastapi import HTTPException from backend.app import config as cfgmod @@ -15,6 +17,61 @@ log = logging.getLogger(__name__) +# ── JWKS reachability probe ───────────────────────────────────────────── +# /api/health is polled by the container healthcheck every 10-30s. Probing +# Supabase on each of those calls means a third-party network round trip on a +# fixed timer forever, and it made the endpoint take 1.8-3.2s. Worse, the old +# code built an httpx.AsyncClient per call and never closed it, so every probe +# leaked its connection pool. +# +# Reachability is not something that changes between two polls seconds apart, +# so cache the answer briefly and share one client. A liveness probe should be +# cheap; the diagnostic value survives a short TTL. +_JWKS_PROBE_TTL_SECONDS = 60.0 +_jwks_probe_cache: tuple[str, str, float] | None = None # (url, status, checked_at) +_jwks_http_client: httpx.AsyncClient | None = None + + +def _get_jwks_http_client() -> httpx.AsyncClient: + global _jwks_http_client + if _jwks_http_client is None or _jwks_http_client.is_closed: + _jwks_http_client = httpx.AsyncClient(timeout=5) + return _jwks_http_client + + +async def close_jwks_http_client() -> None: + """Release the shared probe client on shutdown.""" + global _jwks_http_client, _jwks_probe_cache + if _jwks_http_client is not None and not _jwks_http_client.is_closed: + await _jwks_http_client.aclose() + _jwks_http_client = None + _jwks_probe_cache = None + + +async def _probe_jwks(supabase_url: str) -> str: + """Report whether the Supabase JWKS endpoint answers, cached for a short TTL.""" + global _jwks_probe_cache + + now = time.monotonic() + if _jwks_probe_cache is not None: + cached_url, cached_status, checked_at = _jwks_probe_cache + if cached_url == supabase_url and (now - checked_at) < _JWKS_PROBE_TTL_SECONDS: + return cached_status + + jwks_url = f"{supabase_url}/auth/v1/.well-known/jwks.json" + try: + resp = await _get_jwks_http_client().get(jwks_url) + if resp.status_code == 200: + status = "reachable" + else: + status = f"unexpected_status:{resp.status_code}" + except Exception as e: + status = f"unreachable:{e.__class__.__name__}" + + _jwks_probe_cache = (supabase_url, status, now) + return status + + async def get_state(user_id, workspace_id, db_id, db, cfg, kb): """ Assemble the user's connection, configuration, onboarding, and knowledge state. @@ -88,17 +145,7 @@ async def get_health(pg_status="unknown"): jwks_status = "unchecked" supabase_url = get_supabase_url() if supabase_url: - try: - import httpx - - jwks_url = f"{supabase_url}/auth/v1/.well-known/jwks.json" - resp = await httpx.AsyncClient(timeout=5).get(jwks_url) - if resp.status_code == 200: - jwks_status = "reachable" - else: - jwks_status = f"unexpected_status:{resp.status_code}" - except Exception as e: - jwks_status = f"unreachable:{e.__class__.__name__}" + jwks_status = await _probe_jwks(supabase_url) return { "status": "ok" if pg_status == "connected" else "degraded", diff --git a/backend/app/pgdatabase/engine.py b/backend/app/pgdatabase/engine.py index 94ae005..520fd2c 100644 --- a/backend/app/pgdatabase/engine.py +++ b/backend/app/pgdatabase/engine.py @@ -12,6 +12,23 @@ _session_factory = None +def _env_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None or not raw.strip(): + return default + try: + return int(raw) + except ValueError: + return default + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None or not raw.strip(): + return default + return raw.strip().lower() in ("1", "true", "yes", "on") + + def get_engine(): global _engine if _engine is None: @@ -25,12 +42,33 @@ def get_engine(): db_url = "postgresql+asyncpg://" + db_url[len("postgresql://") :] elif db_url.startswith("postgres://"): db_url = "postgresql+asyncpg://" + db_url[len("postgres://") :] + # ── Pool tuning ── + # pool_pre_ping issues a liveness round trip to the database *before + # handing out every pooled connection*. That is one extra round trip on + # every request that touches the database, and it is not amortised -- + # it repeats even when the connection is demonstrably healthy and was + # used moments ago. Against a managed Postgres it roughly doubled the + # cost of a checkout in our measurements. + # + # What it buys is protection against a connection that died while idle + # in the pool. pool_recycle already closes that window from the other + # side by retiring connections on a timer, well inside any sane server + # or pooler idle timeout, so pre-ping is mostly re-checking what + # recycling has already handled. Default it off and keep recycling. + # + # Every value is env-tunable: a deployment that genuinely sees stale + # connections (an aggressive proxy, a flaky link) can set + # DB_POOL_PRE_PING=true without a code change, and a deployment with + # the database next door can raise the pool instead. _engine = create_async_engine( db_url, - pool_size=5, - max_overflow=10, - pool_recycle=300, - pool_pre_ping=True, + pool_size=_env_int("DB_POOL_SIZE", 5), + max_overflow=_env_int("DB_MAX_OVERFLOW", 10), + pool_recycle=_env_int("DB_POOL_RECYCLE", 300), + pool_pre_ping=_env_bool("DB_POOL_PRE_PING", False), + # Required for pgbouncer-style transaction pooling, which cannot + # carry server-side prepared statements across checkouts. Measured + # as costing nothing here (166ms vs 174ms per query), so it stays. connect_args={"statement_cache_size": 0}, ) return _engine diff --git a/backend/app/server.py b/backend/app/server.py index fe8161b..be4b9d4 100644 --- a/backend/app/server.py +++ b/backend/app/server.py @@ -110,6 +110,10 @@ async def lifespan(app): cleanup_task.cancel() with suppress(asyncio.CancelledError): await cleanup_task + from backend.app.controllers.system import close_jwks_http_client + + with suppress(Exception): + await close_jwks_http_client() await dispose_db() diff --git a/tests/test_engine_pool_config.py b/tests/test_engine_pool_config.py new file mode 100644 index 0000000..48b4e0c --- /dev/null +++ b/tests/test_engine_pool_config.py @@ -0,0 +1,126 @@ +"""Pool configuration is read from the environment, with safe defaults. + +These assert the knobs are actually wired to create_async_engine — a typo in an +env var name would otherwise fail silently and leave the default in place, +which is exactly the kind of thing nobody notices until a deploy is slow. +""" + +import pytest + +from backend.app.pgdatabase import engine as engine_mod + + +@pytest.fixture +def fresh_engine(monkeypatch): + """Capture the kwargs the engine would be built with.""" + captured = {} + + def fake_create_async_engine(url, **kwargs): + captured["url"] = url + captured.update(kwargs) + return object() + + monkeypatch.setattr(engine_mod, "create_async_engine", fake_create_async_engine) + monkeypatch.setattr(engine_mod, "_engine", None) + monkeypatch.setattr( + engine_mod, "_ENGINE_URL", "postgresql://u:p@localhost:5432/testdb" + ) + return captured + + +def test_defaults_disable_pre_ping_and_keep_recycling(fresh_engine, monkeypatch): + for var in ( + "DB_POOL_PRE_PING", + "DB_POOL_RECYCLE", + "DB_POOL_SIZE", + "DB_MAX_OVERFLOW", + ): + monkeypatch.delenv(var, raising=False) + + engine_mod.get_engine() + + # Pre-ping costs a round trip on every checkout; recycling covers the same + # window without that cost, so the default is off-but-recycling. + assert fresh_engine["pool_pre_ping"] is False + assert fresh_engine["pool_recycle"] == 300 + assert fresh_engine["pool_size"] == 5 + assert fresh_engine["max_overflow"] == 10 + + +def test_pre_ping_can_be_re_enabled_without_a_code_change(fresh_engine, monkeypatch): + monkeypatch.setenv("DB_POOL_PRE_PING", "true") + engine_mod.get_engine() + assert fresh_engine["pool_pre_ping"] is True + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("true", True), + ("True", True), + ("1", True), + ("yes", True), + ("on", True), + ("false", False), + ("0", False), + ("no", False), + ("", False), + ("garbage", False), + ], +) +def test_bool_env_parsing(fresh_engine, monkeypatch, raw, expected): + monkeypatch.setenv("DB_POOL_PRE_PING", raw) + engine_mod.get_engine() + assert fresh_engine["pool_pre_ping"] is expected + + +def test_numeric_overrides_are_applied(fresh_engine, monkeypatch): + monkeypatch.setenv("DB_POOL_SIZE", "20") + monkeypatch.setenv("DB_MAX_OVERFLOW", "40") + monkeypatch.setenv("DB_POOL_RECYCLE", "900") + engine_mod.get_engine() + assert fresh_engine["pool_size"] == 20 + assert fresh_engine["max_overflow"] == 40 + assert fresh_engine["pool_recycle"] == 900 + + +def test_unparseable_numbers_fall_back_to_defaults(fresh_engine, monkeypatch): + """A bad value must not take the app down at import time.""" + monkeypatch.setenv("DB_POOL_SIZE", "not-a-number") + engine_mod.get_engine() + assert fresh_engine["pool_size"] == 5 + + +def test_statement_cache_stays_disabled_for_transaction_pooling(fresh_engine): + """pgbouncer transaction mode cannot carry server-side prepared statements.""" + engine_mod.get_engine() + assert fresh_engine["connect_args"] == {"statement_cache_size": 0} + + +@pytest.mark.parametrize( + "given,expected_prefix", + [ + ("postgresql://u:p@h:5432/d", "postgresql+asyncpg://"), + ("postgres://u:p@h:5432/d", "postgresql+asyncpg://"), + ("postgresql+asyncpg://u:p@h:5432/d", "postgresql+asyncpg://"), + ], +) +def test_url_is_normalised_to_asyncpg(monkeypatch, given, expected_prefix): + captured = {} + monkeypatch.setattr( + engine_mod, + "create_async_engine", + lambda url, **kw: captured.update(url=url) or object(), + ) + monkeypatch.setattr(engine_mod, "_engine", None) + monkeypatch.setattr(engine_mod, "_ENGINE_URL", given) + + engine_mod.get_engine() + assert captured["url"].startswith(expected_prefix) + + +def test_missing_url_raises(monkeypatch): + monkeypatch.setattr(engine_mod, "_engine", None) + monkeypatch.setattr(engine_mod, "_ENGINE_URL", None) + with pytest.raises(ValueError, match="DATABASE_URL"): + engine_mod.get_engine() diff --git a/tests/test_health_jwks_probe.py b/tests/test_health_jwks_probe.py new file mode 100644 index 0000000..dca20cc --- /dev/null +++ b/tests/test_health_jwks_probe.py @@ -0,0 +1,140 @@ +"""The /api/health JWKS probe must be cheap, cached, and must not leak clients. + +The container healthcheck polls this endpoint every 10-30s forever, so a probe +that opens a fresh connection pool per call is a slow leak in production. +""" + +import pytest + +from backend.app.controllers import system as sysmod + + +@pytest.fixture(autouse=True) +def reset_probe_state(monkeypatch): + monkeypatch.setattr(sysmod, "_jwks_probe_cache", None) + monkeypatch.setattr(sysmod, "_jwks_http_client", None) + yield + monkeypatch.setattr(sysmod, "_jwks_probe_cache", None) + monkeypatch.setattr(sysmod, "_jwks_http_client", None) + + +class FakeResponse: + def __init__(self, status_code): + self.status_code = status_code + + +class FakeClient: + """Stands in for httpx.AsyncClient, counting requests and closures.""" + + instances = [] + + def __init__(self, *a, **kw): + self.requests = 0 + self.is_closed = False + self.status_code = 200 + self.raises = None + FakeClient.instances.append(self) + + async def get(self, url): + self.requests += 1 + if self.raises: + raise self.raises + return FakeResponse(self.status_code) + + async def aclose(self): + self.is_closed = True + + +@pytest.fixture +def fake_httpx(monkeypatch): + FakeClient.instances = [] + monkeypatch.setattr(sysmod.httpx, "AsyncClient", FakeClient) + return FakeClient + + +@pytest.mark.asyncio +async def test_repeated_probes_reuse_one_client(fake_httpx): + """Every call used to construct — and abandon — its own AsyncClient.""" + for _ in range(5): + sysmod._jwks_probe_cache = None # force a real probe each time + await sysmod._probe_jwks("https://proj.supabase.co") + + assert len(fake_httpx.instances) == 1, "a client was created per probe" + assert fake_httpx.instances[0].requests == 5 + + +@pytest.mark.asyncio +async def test_result_is_cached_within_ttl(fake_httpx): + first = await sysmod._probe_jwks("https://proj.supabase.co") + for _ in range(9): + await sysmod._probe_jwks("https://proj.supabase.co") + + assert first == "reachable" + # Ten calls, one network request: the healthcheck no longer hammers Supabase. + assert fake_httpx.instances[0].requests == 1 + + +@pytest.mark.asyncio +async def test_cache_expires_after_ttl(fake_httpx, monkeypatch): + clock = {"t": 1000.0} + monkeypatch.setattr(sysmod.time, "monotonic", lambda: clock["t"]) + + await sysmod._probe_jwks("https://proj.supabase.co") + clock["t"] += sysmod._JWKS_PROBE_TTL_SECONDS + 1 + await sysmod._probe_jwks("https://proj.supabase.co") + + assert fake_httpx.instances[0].requests == 2 + + +@pytest.mark.asyncio +async def test_cache_is_keyed_on_url(fake_httpx): + await sysmod._probe_jwks("https://one.supabase.co") + await sysmod._probe_jwks("https://two.supabase.co") + assert fake_httpx.instances[0].requests == 2 + + +@pytest.mark.asyncio +async def test_non_200_is_reported_and_cached(fake_httpx): + await sysmod._probe_jwks("https://proj.supabase.co") # creates the client + client = fake_httpx.instances[0] + client.status_code = 503 + sysmod._jwks_probe_cache = None + + assert ( + await sysmod._probe_jwks("https://proj.supabase.co") == "unexpected_status:503" + ) + + +@pytest.mark.asyncio +async def test_network_failure_is_reported_not_raised(fake_httpx): + await sysmod._probe_jwks("https://proj.supabase.co") + client = fake_httpx.instances[0] + client.raises = ConnectionError("boom") + sysmod._jwks_probe_cache = None + + # Health must degrade gracefully, never 500 because a third party is down. + assert ( + await sysmod._probe_jwks("https://proj.supabase.co") + == "unreachable:ConnectionError" + ) + + +@pytest.mark.asyncio +async def test_shutdown_closes_the_client(fake_httpx): + await sysmod._probe_jwks("https://proj.supabase.co") + client = fake_httpx.instances[0] + assert not client.is_closed + + await sysmod.close_jwks_http_client() + assert client.is_closed + assert sysmod._jwks_http_client is None + + +@pytest.mark.asyncio +async def test_client_is_recreated_after_close(fake_httpx): + await sysmod._probe_jwks("https://proj.supabase.co") + await sysmod.close_jwks_http_client() + await sysmod._probe_jwks("https://proj.supabase.co") + + assert len(fake_httpx.instances) == 2 + assert not fake_httpx.instances[1].is_closed