From 012b44d5f2eb69782f7780b80dac2112d8542157 Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Thu, 20 Aug 2026 19:24:39 +0330 Subject: [PATCH] feat(observability): backfill logging for upload, auth, and domain services Why: - resolve_auth_context() runs on every authenticated request and logged nothing; four distinct rejection reasons (malformed/unknown/inactive/ expired key, inactive tenant) were all invisible. - The domain-allowlist rejection in upload_source_file() happens before any ingestion_jobs row exists, so it wasn't covered by the job-level ingestion.job.failed event either -- a rejected upload left zero trace. - Four of upload_source_file()'s five failure branches (parse_failed, chunk_limit_exceeded, embedding_failed, index_failed) called _mark_job_failed(), which wrote to Postgres but never logged; only storage_failed and timeout had an ad-hoc logger.warning duplicated at their own call sites. Changes: - auth/service.py: auth.succeeded / auth.failed (with a reason field per rejection type), matching ADR-0011's own event catalog. - domains/service.py: domain.rejected on the allowlist check; domain.created / domain.updated / domain.status_changed on the three mutations. - files/upload.py: centralized failure logging inside _mark_job_failed (every failure branch already calls it, so logging there once closes all five branches instead of duplicating a log call at each site) as ingestion.job.failed; added ingestion.job.started; renamed the ad-hoc files.upload.succeeded to ingestion.job.completed for catalog consistency. Impact: - None to request/response behavior -- log events only. --- src/application/auth/service.py | 39 +++ src/application/domains/service.py | 28 ++- src/application/files/upload.py | 47 ++-- .../postgres/test_auth_service_logging.py | 134 +++++++++++ .../postgres/test_domains_service_logging.py | 142 +++++++++++ .../postgres/test_upload_service_logging.py | 223 ++++++++++++++++++ 6 files changed, 595 insertions(+), 18 deletions(-) create mode 100644 tests/integration/postgres/test_auth_service_logging.py create mode 100644 tests/integration/postgres/test_domains_service_logging.py create mode 100644 tests/integration/postgres/test_upload_service_logging.py diff --git a/src/application/auth/service.py b/src/application/auth/service.py index 93bdc36..22c223e 100644 --- a/src/application/auth/service.py +++ b/src/application/auth/service.py @@ -8,6 +8,7 @@ must run with no Postgres session held open at all. from datetime import UTC, datetime +import structlog from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.auth.context import AuthContext @@ -16,28 +17,66 @@ from src.application.auth.keys import parse_api_key, verify_secret from src.infrastructure.postgres.repositories import api_keys as api_keys_repo from src.infrastructure.postgres.repositories import tenants as tenants_repo +logger = structlog.get_logger(__name__) + async def resolve_auth_context( sessionmaker: async_sessionmaker[AsyncSession], bearer_token: str ) -> AuthContext: + """Resolve a bearer token, logging the outcome either way (ADR-0011). + + This runs on every authenticated request, so `auth.failed` is the one + event most likely to matter first when diagnosing a client integration + issue -- and the reason string alone (never logged; it can echo back + attacker-supplied key material) is not enough to tell a malformed token + apart from a revoked one without this. + """ parsed = parse_api_key(bearer_token) if parsed is None: + logger.warning("auth.failed", reason="malformed_key") raise InvalidApiKeyError("malformed API key") key_prefix, secret = parsed async with sessionmaker() as session: api_key = await api_keys_repo.get_by_prefix(session, key_prefix) if api_key is None or not verify_secret(secret, api_key.key_hash): + logger.warning("auth.failed", reason="unknown_key", key_prefix=key_prefix) raise InvalidApiKeyError("unknown API key") if api_key.status != "active": + logger.warning( + "auth.failed", + reason="key_inactive", + key_prefix=key_prefix, + api_key_id=str(api_key.id), + key_status=api_key.status, + ) raise InvalidApiKeyError(f"API key is {api_key.status}") if api_key.expires_at is not None and api_key.expires_at <= datetime.now(UTC): + logger.warning( + "auth.failed", + reason="key_expired", + key_prefix=key_prefix, + api_key_id=str(api_key.id), + ) raise InvalidApiKeyError("API key has expired") tenant = await tenants_repo.get_by_id(session, api_key.tenant_id) if tenant is None or tenant.status != "active": + logger.warning( + "auth.failed", + reason="tenant_inactive", + key_prefix=key_prefix, + api_key_id=str(api_key.id), + tenant_id=str(api_key.tenant_id), + ) raise TenantInactiveError("tenant is not active") + logger.info( + "auth.succeeded", + tenant_id=str(tenant.id), + api_key_id=str(api_key.id), + actor_type=api_key.actor_type, + ) return AuthContext( tenant_id=tenant.id, tenant_slug=tenant.slug, diff --git a/src/application/domains/service.py b/src/application/domains/service.py index 90eba71..681a76b 100644 --- a/src/application/domains/service.py +++ b/src/application/domains/service.py @@ -15,6 +15,7 @@ a request body (ADR-0002). import uuid +import structlog from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError @@ -22,6 +23,8 @@ from src.application.domains.models import DomainResult from src.infrastructure.postgres.models.tenant_domain import TenantDomain from src.infrastructure.postgres.repositories import tenant_domains as repo +logger = structlog.get_logger(__name__) + async def _flush_and_refresh(session: AsyncSession, tenant_domain: TenantDomain) -> None: """Materialize server-generated columns before the row leaves the session. @@ -55,14 +58,25 @@ async def ensure_domain_allowed( Takes a session rather than a sessionmaker: the upload path calls this inside its existing txn A, so the check costs no extra connection and cannot pass and then go stale before the row is written. + + Logs the rejection here rather than at the call site: this runs before any + `ingestion_jobs` row exists, so `upload_source_file`'s job-level + `ingestion.job.failed` event (ADR-0011) never fires for it -- without a log + here, a rejected upload would leave no operational trace at all. """ tenant_domain = await repo.get(session, tenant_id=tenant_id, domain=domain) if tenant_domain is None: + logger.warning( + "domain.rejected", tenant_id=str(tenant_id), domain=domain, reason="unregistered" + ) raise UnknownDomainError( f"domain '{domain}' is not registered for this tenant; " f"create it via POST /v1/domains before uploading to it" ) if tenant_domain.status != "active": + logger.warning( + "domain.rejected", tenant_id=str(tenant_id), domain=domain, reason="disabled" + ) raise UnknownDomainError(f"domain '{domain}' is disabled for this tenant") @@ -98,7 +112,9 @@ async def create_domain( metadata=metadata, ) await session.commit() - return _to_result(created) + + logger.info("domain.created", tenant_id=str(tenant_id), domain=domain) + return _to_result(created) async def update_domain( @@ -116,7 +132,10 @@ async def update_domain( repo.update_display_name(found, display_name=display_name) await _flush_and_refresh(session, found) await session.commit() - return _to_result(found) + result = _to_result(found) + + logger.info("domain.updated", tenant_id=str(tenant_id), domain=domain) + return result async def set_domain_status( @@ -139,4 +158,7 @@ async def set_domain_status( repo.set_status(found, status=status) await _flush_and_refresh(session, found) await session.commit() - return _to_result(found) + result = _to_result(found) + + logger.info("domain.status_changed", tenant_id=str(tenant_id), domain=domain, status=status) + return result diff --git a/src/application/files/upload.py b/src/application/files/upload.py index fb63214..3c54a5f 100644 --- a/src/application/files/upload.py +++ b/src/application/files/upload.py @@ -68,6 +68,16 @@ async def _mark_job_failed( error_code: str, error_message: str, ) -> None: + """Write the terminal `failed` job row and emit its log event together. + + Every failure branch below calls this, so logging here once closes every + branch at once rather than duplicating a `logger.warning` at each call + site (CLAUDE.md, "prefer deep modules") -- previously only + `storage_upload_failed` and `timeout` did that ad hoc, and + `parse_failed`/`chunk_limit_exceeded`/`embedding_failed`/`index_failed` + logged nothing at all: visible in `ingestion_job_events` but invisible to + log-based alerting (ADR-0011). + """ async with sessionmaker() as session: job = await jobs_repo.mark_terminal( session, @@ -88,6 +98,14 @@ async def _mark_job_failed( ) await session.commit() + logger.warning( + "ingestion.job.failed", + tenant_id=str(tenant_id), + ingestion_job_id=str(ingestion_job_id), + error_code=error_code, + error_message=error_message, + ) + async def upload_source_file( *, @@ -191,6 +209,15 @@ async def upload_source_file( await session.commit() ingestion_job_id = job.id + logger.info( + "ingestion.job.started", + tenant_id=str(auth.tenant_id), + ingestion_job_id=str(ingestion_job_id), + file_id=str(source_file_id), + domain=domain, + source_type=validated.source_type, + ) + # Phase 2: no Postgres session open across this work (ADR-0017), # bounded end-to-end by INGESTION_TIMEOUT_SECONDS. try: @@ -200,12 +227,6 @@ async def upload_source_file( key=object_key, data=data, content_type=validated.content_type ) except Exception as exc: - logger.warning( - "files.upload.storage_failed", - tenant_id=str(auth.tenant_id), - file_id=str(source_file_id), - ingestion_job_id=str(ingestion_job_id), - ) await _mark_job_failed( sessionmaker, tenant_id=auth.tenant_id, @@ -288,12 +309,6 @@ async def upload_source_file( ) raise except TimeoutError: - logger.warning( - "files.upload.timeout", - tenant_id=str(auth.tenant_id), - file_id=str(source_file_id), - ingestion_job_id=str(ingestion_job_id), - ) await _mark_job_failed( sessionmaker, tenant_id=auth.tenant_id, @@ -334,11 +349,13 @@ async def upload_source_file( await session.commit() logger.info( - "files.upload.succeeded", + "ingestion.job.completed", tenant_id=str(auth.tenant_id), - file_id=str(source_file_id), ingestion_job_id=str(ingestion_job_id), - points_indexed=indexed.points_upserted, + file_id=str(source_file_id), + chunks_parsed=len(chunks), + points_upserted=indexed.points_upserted, + points_soft_deleted=indexed.points_soft_deleted, ) return UploadResult( file_id=source_file_id, diff --git a/tests/integration/postgres/test_auth_service_logging.py b/tests/integration/postgres/test_auth_service_logging.py new file mode 100644 index 0000000..29c4d7d --- /dev/null +++ b/tests/integration/postgres/test_auth_service_logging.py @@ -0,0 +1,134 @@ +"""`resolve_auth_context` emits `auth.succeeded`/`auth.failed` (ADR-0011). + +This runs on every authenticated request, so every rejection reason needs a +distinguishable log event -- previously none of them logged anything. +""" + +from collections.abc import MutableMapping +from typing import Any + +import pytest +import structlog +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError +from src.application.auth.service import resolve_auth_context +from tests.support.factories import create_api_key, create_tenant + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +def _events_by_name( + logs: list[MutableMapping[str, Any]], name: str +) -> list[MutableMapping[str, Any]]: + return [entry for entry in logs if entry.get("event") == name] + + +async def test_valid_key_emits_auth_succeeded( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + _, full_key = await create_api_key(db_session, tenant=tenant, scopes=["files:write"]) + await db_session.commit() + + with structlog.testing.capture_logs() as logs: + auth = await resolve_auth_context(db_sessionmaker, full_key) + + succeeded = _events_by_name(logs, "auth.succeeded") + assert len(succeeded) == 1 + assert succeeded[0]["tenant_id"] == str(auth.tenant_id) + assert succeeded[0]["api_key_id"] == str(auth.api_key_id) + assert _events_by_name(logs, "auth.failed") == [] + + +async def test_malformed_token_emits_auth_failed( + db_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + with structlog.testing.capture_logs() as logs, pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, "not-a-bearer-token-at-all") + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert failed[0]["reason"] == "malformed_key" + + +async def test_wrong_secret_emits_auth_failed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + + with structlog.testing.capture_logs() as logs, pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_wrong-secret") + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert failed[0]["reason"] == "unknown_key" + + +async def test_unknown_prefix_emits_auth_failed( + db_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + with structlog.testing.capture_logs() as logs, pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, "sk_doesnotexist_secret") + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert failed[0]["reason"] == "unknown_key" + + +async def test_revoked_key_emits_auth_failed_with_key_status( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + _, full_key = await create_api_key(db_session, tenant=tenant, status="revoked") + await db_session.commit() + + with structlog.testing.capture_logs() as logs, pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, full_key) + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert failed[0]["reason"] == "key_inactive" + assert failed[0]["key_status"] == "revoked" + + +async def test_suspended_tenant_emits_auth_failed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session, status="suspended") + _, full_key = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + + with structlog.testing.capture_logs() as logs, pytest.raises(TenantInactiveError): + await resolve_auth_context(db_sessionmaker, full_key) + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert failed[0]["reason"] == "tenant_inactive" + assert failed[0]["tenant_id"] == str(tenant.id) + + +async def test_auth_failed_never_logs_the_secret( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """ADR-0011's redaction rule: never log plaintext API keys. `key_prefix` + is the non-secret lookup portion (same distinction `ApiKey.key_prefix` + makes); the secret itself must not appear in any field's value. + """ + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + wrong_secret = "definitely-not-the-real-secret" + + with structlog.testing.capture_logs() as logs, pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_{wrong_secret}") + + failed = _events_by_name(logs, "auth.failed") + assert len(failed) == 1 + assert wrong_secret not in str(failed[0]) diff --git a/tests/integration/postgres/test_domains_service_logging.py b/tests/integration/postgres/test_domains_service_logging.py new file mode 100644 index 0000000..f91bcf5 --- /dev/null +++ b/tests/integration/postgres/test_domains_service_logging.py @@ -0,0 +1,142 @@ +"""`domains/service.py` emits log events for the allowlist rejection and every +mutation (ADR-0011). `ensure_domain_allowed` is the one that matters most: it +runs before any `ingestion_jobs` row exists, so without its own log a rejected +upload leaves no operational trace at all. +""" + +from collections.abc import MutableMapping +from typing import Any + +import pytest +import structlog +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.domains import ( + DomainAlreadyExistsError, + UnknownDomainError, + create_domain, + ensure_domain_allowed, + set_domain_status, + update_domain, +) +from tests.support.factories import create_tenant, create_tenant_domain + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +def _events_by_name( + logs: list[MutableMapping[str, Any]], name: str +) -> list[MutableMapping[str, Any]]: + return [entry for entry in logs if entry.get("event") == name] + + +async def test_ensure_domain_allowed_logs_nothing_when_the_domain_is_active( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + + with structlog.testing.capture_logs() as logs: + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fire") + + assert _events_by_name(logs, "domain.rejected") == [] + + +async def test_ensure_domain_allowed_logs_rejection_for_an_unregistered_domain( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + + with structlog.testing.capture_logs() as logs, pytest.raises(UnknownDomainError): + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fier") + + rejected = _events_by_name(logs, "domain.rejected") + assert len(rejected) == 1 + assert rejected[0]["reason"] == "unregistered" + assert rejected[0]["domain"] == "fier" + assert rejected[0]["tenant_id"] == str(tenant.id) + + +async def test_ensure_domain_allowed_logs_rejection_for_a_disabled_domain( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire", status="disabled") + + with structlog.testing.capture_logs() as logs, pytest.raises(UnknownDomainError): + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fire") + + rejected = _events_by_name(logs, "domain.rejected") + assert len(rejected) == 1 + assert rejected[0]["reason"] == "disabled" + + +async def test_create_domain_emits_domain_created( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await db_session.commit() + + with structlog.testing.capture_logs() as logs: + await create_domain( + db_sessionmaker, tenant_id=tenant.id, domain="car", display_name="Car insurance" + ) + + created = _events_by_name(logs, "domain.created") + assert len(created) == 1 + assert created[0]["domain"] == "car" + + +async def test_create_domain_duplicate_does_not_emit_domain_created( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + with structlog.testing.capture_logs() as logs, pytest.raises(DomainAlreadyExistsError): + await create_domain( + db_sessionmaker, tenant_id=tenant.id, domain="fire", display_name="Fire again" + ) + + assert _events_by_name(logs, "domain.created") == [] + + +async def test_update_domain_emits_domain_updated( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + with structlog.testing.capture_logs() as logs: + await update_domain( + db_sessionmaker, tenant_id=tenant.id, domain="fire", display_name="Fire & perils" + ) + + updated = _events_by_name(logs, "domain.updated") + assert len(updated) == 1 + assert updated[0]["domain"] == "fire" + + +async def test_set_domain_status_emits_domain_status_changed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + with structlog.testing.capture_logs() as logs: + await set_domain_status( + db_sessionmaker, tenant_id=tenant.id, domain="fire", status="disabled" + ) + + changed = _events_by_name(logs, "domain.status_changed") + assert len(changed) == 1 + assert changed[0]["domain"] == "fire" + assert changed[0]["status"] == "disabled" diff --git a/tests/integration/postgres/test_upload_service_logging.py b/tests/integration/postgres/test_upload_service_logging.py new file mode 100644 index 0000000..518cd1d --- /dev/null +++ b/tests/integration/postgres/test_upload_service_logging.py @@ -0,0 +1,223 @@ +"""`upload_source_file` emits `ingestion.job.*` log events (ADR-0011). + +Every failure branch funnels through `_mark_job_failed`, so this asserts the +log event once per branch rather than re-testing the Postgres job-row +behavior already covered in `test_upload_service.py`. Uses +`structlog.testing.capture_logs()`, which captures events independent of +whichever handlers/renderers happen to be configured in this process. +""" + +from collections.abc import MutableMapping +from typing import Any + +import pytest +import structlog +from anyio import CapacityLimiter, Semaphore +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.context import AuthContext +from src.application.files.upload import upload_source_file +from src.application.ingestion.errors import ( + EmbedderError, + IngestionTimeoutError, + PointIndexingError, +) +from src.config import ChunkingSettings, IngestionSettings, QdrantSettings +from tests.fakes import ( + FakeDenseEmbedder, + FakeObjectStorage, + FakePointStorage, + FakeSparseEmbedder, +) +from tests.support.factories import create_api_key, create_tenant, create_tenant_domain + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + +_CSV_BYTES = b"name,value\nfirst,1\n" + + +async def _auth_for(db_session: AsyncSession, *, domain: str = "general") -> AuthContext: + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await create_tenant_domain(db_session, tenant=tenant, domain=domain) + await db_session.commit() + return AuthContext( + tenant_id=tenant.id, + tenant_slug=tenant.slug, + api_key_id=api_key.id, + scopes=frozenset({"files:write"}), + actor_type="backend", + ) + + +def _events_by_name( + logs: list[MutableMapping[str, Any]], name: str +) -> list[MutableMapping[str, Any]]: + return [entry for entry in logs if entry.get("event") == name] + + +async def test_successful_upload_emits_started_and_completed_events( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + + with structlog.testing.capture_logs() as logs: + result = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + point_storage=FakePointStorage(), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + ingestion_settings=IngestionSettings(), + chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + concurrency_limiter=Semaphore(2), + dense_embedders=[ + FakeDenseEmbedder(name="dense_nomic"), + FakeDenseEmbedder(name="dense_openai"), + ], + sparse_embedder=FakeSparseEmbedder(), + ) + + started = _events_by_name(logs, "ingestion.job.started") + completed = _events_by_name(logs, "ingestion.job.completed") + assert len(started) == 1 + assert started[0]["tenant_id"] == str(auth.tenant_id) + assert started[0]["ingestion_job_id"] == str(result.ingestion_job_id) + assert len(completed) == 1 + assert completed[0]["points_upserted"] == result.chunks_indexed + assert _events_by_name(logs, "ingestion.job.failed") == [] + + +async def test_storage_failure_emits_ingestion_job_failed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + + with structlog.testing.capture_logs() as logs, pytest.raises(OSError): + await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(fail_next=True), + point_storage=FakePointStorage(), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + ingestion_settings=IngestionSettings(), + chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + concurrency_limiter=Semaphore(2), + dense_embedders=[ + FakeDenseEmbedder(name="dense_nomic"), + FakeDenseEmbedder(name="dense_openai"), + ], + sparse_embedder=FakeSparseEmbedder(), + ) + + failed = _events_by_name(logs, "ingestion.job.failed") + assert len(failed) == 1 + assert failed[0]["error_code"] == "storage_upload_failed" + assert failed[0]["tenant_id"] == str(auth.tenant_id) + + +async def test_embedding_failure_emits_ingestion_job_failed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """Previously silent: embedding_failed reached Postgres but never logged.""" + auth = await _auth_for(db_session) + failing_embedder = FakeDenseEmbedder(name="dense_nomic", fail_next=True) + + with structlog.testing.capture_logs() as logs, pytest.raises(EmbedderError): + await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + point_storage=FakePointStorage(), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + ingestion_settings=IngestionSettings(), + chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + concurrency_limiter=Semaphore(2), + dense_embedders=[failing_embedder, FakeDenseEmbedder(name="dense_openai")], + sparse_embedder=FakeSparseEmbedder(), + ) + + failed = _events_by_name(logs, "ingestion.job.failed") + assert len(failed) == 1 + assert failed[0]["error_code"] == "embedding_failed" + + +async def test_index_failure_emits_ingestion_job_failed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """Previously silent: index_failed reached Postgres but never logged.""" + auth = await _auth_for(db_session) + + with structlog.testing.capture_logs() as logs, pytest.raises(PointIndexingError): + await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + point_storage=FakePointStorage(fail_on_batch=0), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + ingestion_settings=IngestionSettings(), + chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + concurrency_limiter=Semaphore(2), + dense_embedders=[ + FakeDenseEmbedder(name="dense_nomic"), + FakeDenseEmbedder(name="dense_openai"), + ], + sparse_embedder=FakeSparseEmbedder(), + ) + + failed = _events_by_name(logs, "ingestion.job.failed") + assert len(failed) == 1 + assert failed[0]["error_code"] == "index_failed" + + +async def test_timeout_emits_ingestion_job_failed_not_a_duplicate_event( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """The old ad-hoc `files.upload.timeout` log is gone -- `_mark_job_failed` + is now the single place a failure is logged, so there is exactly one + `ingestion.job.failed` event, not two events for one failure. + """ + auth = await _auth_for(db_session) + slow_embedder = FakeDenseEmbedder(name="dense_nomic", delay_seconds=10) + + with structlog.testing.capture_logs() as logs, pytest.raises(IngestionTimeoutError): + await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + point_storage=FakePointStorage(), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + ingestion_settings=IngestionSettings(timeout_seconds=0.05), + chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), + thread_limiter=CapacityLimiter(2), + concurrency_limiter=Semaphore(2), + dense_embedders=[slow_embedder, FakeDenseEmbedder(name="dense_openai")], + sparse_embedder=FakeSparseEmbedder(), + ) + + failed = _events_by_name(logs, "ingestion.job.failed") + assert len(failed) == 1 + assert failed[0]["error_code"] == "timeout"