diff --git a/.env.example b/.env.example index 29edaf8..95cc871 100644 --- a/.env.example +++ b/.env.example @@ -113,6 +113,10 @@ TAIL_ERROR_THRESHOLD=5 # WEBHOOK_MAX_RETRIES=5 # WEBHOOK_TIMEOUT=10 +# Idempotency-Key TTL for POST /v1/ingestions (batch enqueue and tail create). +# Repeat requests with the same key within this window return the original job. +# INGEST_IDEMPOTENCY_TTL_SECONDS=86400 + # ── Adapters ────────────────────────────────────────────────────────────────── # AWS region for the cloudwatch source adapter (raglogs ingest --adapter cloudwatch). diff --git a/README.md b/README.md index 95dc3b1..db19757 100644 --- a/README.md +++ b/README.md @@ -688,6 +688,7 @@ All settings are read from `.env`, environment variables, or CLI flags. Priority | `WEBHOOK_SECRET` | _(empty)_ | Fallback HMAC secret for ingest completion callbacks when auth is off or the API key has no per-key `whsec_` | | `WEBHOOK_MAX_RETRIES` | `5` | Extra webhook POST attempts after the first (6 POSTs by default) on 5xx / 429 / connect errors | | `WEBHOOK_TIMEOUT` | `10` | Per-attempt HTTP timeout in seconds for completion callbacks | +| `INGEST_IDEMPOTENCY_TTL_SECONDS` | `86400` | How long `Idempotency-Key` on `POST /v1/ingestions` is remembered (batch enqueue and tail create) | --- @@ -993,6 +994,17 @@ curl -X POST http://localhost:8000/v1/ingestions/$ID:stop **Backpressure.** When pending worker jobs ≥ `INGEST_QUEUE_MAX` (default 100), `POST /v1/ingestions` and `POST /v1/ingestions/lines` return **429** with `Retry-After` (`INGEST_RETRY_AFTER_SECONDS`, default 5) and body `{"error_code":"INGEST_QUEUE_FULL","message":"..."}`. This is a queue-depth stand-in, not full API/LLM rate limiting. +**Idempotency-Key.** `POST /v1/ingestions` (batch enqueue and tail create; also the deprecated `/ingestions` alias) honors an `Idempotency-Key` header (max 256 characters). A repeat within `INGEST_IDEMPOTENCY_TTL_SECONDS` (default 86400) returns the original **202** job — the same `worker_job_id` for batch, the same `ingestion_job_id` for tail — instead of starting a new one. Empty keys return **400**. GET routes ignore the header. `POST /v1/ingestions/lines` does not use the header; duplicate push/tail lines are handled by content dedup instead. + +```bash +curl -X POST http://localhost:8000/v1/ingestions \ + -H "Content-Type: application/json" \ + -H "Idempotency-Key: incident-123-retry" \ + -d '{"paths":["/var/log/app"]}' +``` + +**Content dedup.** Every persist path (`ingest_files`, `ingest_from_source` including tail ticks, and push `/lines`) stores `original_line_hash` (SHA-256 of the **raw** line, distinct from the normalized fingerprint) and upserts on `(scope, source_ref, original_line_hash, timestamp)`. Re-reading the same physical lines is a no-op, so cluster counts stay stable across overlapping windows and tail/push retries. Missing `source_ref` is stored as `""` so uniqueness works (Postgres NULLs are distinct). `scope` defaults to `"default"`, or the API key's scope when a principal is present; queries are **not** filtered by scope yet. Duplicate lines are skipped, not errors. + **Completion callbacks.** Optional `callback_url` on `POST /v1/ingestions` (http or https only; `file:` and empty hosts are rejected). When a **batch** worker job reaches a terminal state (`done` / `failed`), raglogs POSTs an HMAC-SHA256-signed JSON body to that URL. Delivery is fail-open: retries with jittered exponential backoff (`WEBHOOK_MAX_RETRIES`, default 5 extra attempts) on 5xx, 429, and connect errors; 4xx other than 429 are not retried. Failures are logged and **do not** change ingest status — poll `GET /v1/ingestions/jobs/{worker_job_id}` still works. `job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the authenticated key's scope, or `"default"` when auth is off. `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`. diff --git a/clients/openapi.json b/clients/openapi.json index 86cf0e8..3ab2c83 100644 --- a/clients/openapi.json +++ b/clients/openapi.json @@ -28,42 +28,42 @@ } }, "/v1/ingestions": { - "get": { - "tags": [ - "ingestions" - ], - "summary": "List Ingestions", - "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", - "operationId": "v1_ingestions_list_ingestions__v1_ingestions", - "responses": { - "200": { - "description": "Successful Response", - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/IngestionListResponse" - } - } - } - } - } - }, "post": { "tags": [ "ingestions" ], "summary": "Create Ingestion", - "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).", + "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).\n\nOptional ``Idempotency-Key`` (also accepted as ``idempotency-key``): a\nrepeat within the TTL returns the original 202 job rather than a new one.", "operationId": "v1_ingestions_create_ingestion__v1_ingestions", + "parameters": [ + { + "name": "Idempotency-Key", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "If set, a repeat POST within INGEST_IDEMPOTENCY_TTL_SECONDS (default 86400) returns the original job instead of enqueueing a new one. Empty values are rejected with 400. Applies to batch enqueue and tail create.", + "title": "Idempotency-Key" + }, + "description": "If set, a repeat POST within INGEST_IDEMPOTENCY_TTL_SECONDS (default 86400) returns the original job instead of enqueueing a new one. Empty values are rejected with 400. Applies to batch enqueue and tail create." + } + ], "requestBody": { + "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/IngestRequest" } } - }, - "required": true + } }, "responses": { "202": { @@ -87,6 +87,26 @@ } } } + }, + "get": { + "tags": [ + "ingestions" + ], + "summary": "List Ingestions", + "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", + "operationId": "v1_ingestions_list_ingestions__v1_ingestions", + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/IngestionListResponse" + } + } + } + } + } } }, "/v1/ingestions/jobs/{worker_job_id}": { @@ -366,43 +386,43 @@ } }, "/ingestions": { - "get": { - "tags": [ - "ingestions" - ], - "summary": "List Ingestions", - "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", - "operationId": "legacy_ingestions_list_ingestions__ingestions", - "responses": { - "200": { - "description": "Successful Response", - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/IngestionListResponse" - } - } - } - } - }, - "deprecated": true - }, "post": { "tags": [ "ingestions" ], "summary": "Create Ingestion", - "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).", + "description": "Enqueue an ingest job. Returns immediately with worker_job_id.\nPoll GET /ingestions/jobs/{worker_job_id} for progress.\nWhen done, use ingestion_job_id from the result to scope /query/explain.\n\n``mode=tail`` creates a long-lived job that the worker polls; the response\nincludes ``ingestion_job_id`` and no worker_job_id.\n\nOptional ``callback_url`` (http/https) is stored on batch worker jobs. The\nworker POSTs an HMAC-signed completion payload on terminal state. Tail jobs\ndo not fire callbacks (long-lived; poll or ``:stop`` instead).\n\nOptional ``Idempotency-Key`` (also accepted as ``idempotency-key``): a\nrepeat within the TTL returns the original 202 job rather than a new one.", "operationId": "legacy_ingestions_create_ingestion__ingestions", + "deprecated": true, + "parameters": [ + { + "name": "Idempotency-Key", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "If set, a repeat POST within INGEST_IDEMPOTENCY_TTL_SECONDS (default 86400) returns the original job instead of enqueueing a new one. Empty values are rejected with 400. Applies to batch enqueue and tail create.", + "title": "Idempotency-Key" + }, + "description": "If set, a repeat POST within INGEST_IDEMPOTENCY_TTL_SECONDS (default 86400) returns the original job instead of enqueueing a new one. Empty values are rejected with 400. Applies to batch enqueue and tail create." + } + ], "requestBody": { + "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/IngestRequest" } } - }, - "required": true + } }, "responses": { "202": { @@ -425,8 +445,28 @@ } } } - }, - "deprecated": true + } + }, + "get": { + "tags": [ + "ingestions" + ], + "summary": "List Ingestions", + "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", + "operationId": "legacy_ingestions_list_ingestions__ingestions", + "deprecated": true, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/IngestionListResponse" + } + } + } + } + } } }, "/ingestions/jobs/{worker_job_id}": { diff --git a/migrations/versions/0007_ingest_idempotency.py b/migrations/versions/0007_ingest_idempotency.py new file mode 100644 index 0000000..df42b35 --- /dev/null +++ b/migrations/versions/0007_ingest_idempotency.py @@ -0,0 +1,78 @@ +"""ingest idempotency keys and log-entry content dedup + +Revision ID: 0007_ingest_idempotency +Revises: 0006_api_key_webhook_secret +Create Date: 2026-08-17 00:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision: str = "0007_ingest_idempotency" +down_revision: Union[str, None] = "0006_api_key_webhook_secret" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "log_entries", + sa.Column("original_line_hash", sa.String(64), nullable=True), + ) + op.add_column( + "log_entries", + sa.Column("scope", sa.String(255), nullable=False, server_default="default"), + ) + op.create_index( + "ux_log_entries_dedup", + "log_entries", + ["scope", "source_ref", "original_line_hash", "timestamp"], + unique=True, + postgresql_where=sa.text( + "original_line_hash IS NOT NULL AND timestamp IS NOT NULL " + "AND source_ref IS NOT NULL" + ), + ) + op.create_table( + "ingest_idempotency_keys", + sa.Column("key", sa.String(256), primary_key=True), + sa.Column( + "worker_job_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("worker_jobs.id"), + nullable=True, + ), + sa.Column( + "ingestion_job_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("ingestion_jobs.id"), + nullable=True, + ), + sa.Column("mode", sa.String(20), nullable=False, server_default="batch"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + ), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index( + "ix_ingest_idempotency_keys_expires_at", + "ingest_idempotency_keys", + ["expires_at"], + ) + + +def downgrade() -> None: + op.drop_index( + "ix_ingest_idempotency_keys_expires_at", + table_name="ingest_idempotency_keys", + ) + op.drop_table("ingest_idempotency_keys") + op.drop_index("ux_log_entries_dedup", table_name="log_entries") + op.drop_column("log_entries", "scope") + op.drop_column("log_entries", "original_line_hash") diff --git a/src/api/routes/ingestions.py b/src/api/routes/ingestions.py index 6b49f53..849e5d6 100644 --- a/src/api/routes/ingestions.py +++ b/src/api/routes/ingestions.py @@ -14,7 +14,7 @@ from datetime import datetime, timezone from typing import Any, Literal, Optional -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, Header, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, field_validator, model_validator @@ -238,10 +238,11 @@ def _ingest_payload( "with_embeddings": request.with_embeddings, "mode": request.mode, } + if callback_meta: + # scope / api_key_id belong on every ingest, not only webhook deliveries. + payload.update(callback_meta) if request.callback_url: payload["callback_url"] = request.callback_url - if callback_meta: - payload.update(callback_meta) return payload @@ -260,6 +261,80 @@ def _callback_meta_from_request(http_request: Request) -> dict[str, Any]: return meta +def _scope_from_request(http_request: Request) -> str: + return str(_callback_meta_from_request(http_request).get("scope") or "default") + + +def _enqueued_from_idempotency(row: Any) -> EnqueuedResponse: + mode = getattr(row, "mode", None) or "batch" + ingestion_job_id = getattr(row, "ingestion_job_id", None) + worker_job_id = getattr(row, "worker_job_id", None) + if mode == "tail": + return EnqueuedResponse( + worker_job_id=None, + status="running", + ingestion_job_id=str(ingestion_job_id) if ingestion_job_id else None, + mode="tail", + ) + return EnqueuedResponse( + worker_job_id=str(worker_job_id) if worker_job_id else None, + status="pending", + ingestion_job_id=str(ingestion_job_id) if ingestion_job_id else None, + mode="batch", + ) + + +def _replay_if_active(db: Any, key: Optional[str]) -> Optional[EnqueuedResponse]: + if not key: + return None + from datetime import datetime, timezone + + from src.core.ingestion.idempotency import key_log_prefix, lookup_active_key + + row = lookup_active_key(db, key, datetime.now(tz=timezone.utc)) + if row is None: + return None + import structlog + + structlog.get_logger().info( + "ingest_idempotency_replay", + key_prefix=key_log_prefix(key), + mode=getattr(row, "mode", None) or "batch", + ) + return _enqueued_from_idempotency(row) + + +def _bind_idempotency_key( + db: Any, + key: Optional[str], + *, + worker_job_id: Optional[uuid.UUID], + ingestion_job_id: Optional[uuid.UUID], + mode: str, +) -> Optional[EnqueuedResponse]: + """Persist the key mapping. Returns a replay response on unique-key race.""" + if not key: + return None + from datetime import datetime, timezone + + from src.config import get_settings + from src.core.ingestion.idempotency import store_idempotency_key + + settings = get_settings() + _row, is_replay = store_idempotency_key( + db, + key, + worker_job_id=worker_job_id, + ingestion_job_id=ingestion_job_id, + mode=mode, + now=datetime.now(tz=timezone.utc), + ttl_seconds=settings.ingest_idempotency_ttl_seconds, + ) + if is_replay: + return _enqueued_from_idempotency(_row) + return None + + def _create_tail_job( request: IngestRequest, params: dict[str, Any], @@ -331,7 +406,19 @@ def _apply_tail_action( @router.post("", response_model=EnqueuedResponse, status_code=202) -def create_ingestion(request: IngestRequest, http_request: Request): +def create_ingestion( + request: IngestRequest, + http_request: Request, + idempotency_key: Optional[str] = Header( + default=None, + alias="Idempotency-Key", + description=( + "If set, a repeat POST within INGEST_IDEMPOTENCY_TTL_SECONDS " + "(default 86400) returns the original job instead of enqueueing a new one. " + "Empty values are rejected with 400. Applies to batch enqueue and tail create." + ), + ), +): """ Enqueue an ingest job. Returns immediately with worker_job_id. Poll GET /ingestions/jobs/{worker_job_id} for progress. @@ -343,10 +430,22 @@ def create_ingestion(request: IngestRequest, http_request: Request): Optional ``callback_url`` (http/https) is stored on batch worker jobs. The worker POSTs an HMAC-signed completion payload on terminal state. Tail jobs do not fire callbacks (long-lived; poll or ``:stop`` instead). + + Optional ``Idempotency-Key`` (also accepted as ``idempotency-key``): a + repeat within the TTL returns the original 202 job rather than a new one. """ + from src.core.ingestion.idempotency import ( + InvalidIdempotencyKey, + parse_idempotency_key, + ) from src.db.models import WorkerJob from src.db.session import get_db + try: + parsed_key = parse_idempotency_key(idempotency_key) + except InvalidIdempotencyKey as e: + raise HTTPException(status_code=400, detail=str(e)) from e + params = request.params if request.adapter == "file": from src.adapters.file.adapter import discover_files @@ -362,12 +461,41 @@ def create_ingestion(request: IngestRequest, http_request: Request): callback_meta = _callback_meta_from_request(http_request) with get_db() as db: + replayed = _replay_if_active(db, parsed_key) + if replayed is not None: + return replayed + rejected = _reject_if_queue_full(db) if rejected is not None: return rejected if request.mode == "tail": - return _create_tail_job(request, params, db, callback_meta=callback_meta) + response = _create_tail_job( + request, params, db, callback_meta=callback_meta + ) + race = _bind_idempotency_key( + db, + parsed_key, + worker_job_id=None, + ingestion_job_id=uuid.UUID(response.ingestion_job_id) + if response.ingestion_job_id + else None, + mode="tail", + ) + if race is not None: + from src.db.models import IngestionJob + + if response.ingestion_job_id: + orphan = ( + db.query(IngestionJob) + .filter(IngestionJob.id == uuid.UUID(response.ingestion_job_id)) + .first() + ) + if orphan is not None: + db.delete(orphan) + db.flush() + return race + return response payload = _ingest_payload(request, params, callback_meta=callback_meta) job = WorkerJob( @@ -379,6 +507,17 @@ def create_ingestion(request: IngestRequest, http_request: Request): db.add(job) db.flush() worker_job_id = str(job.id) + race = _bind_idempotency_key( + db, + parsed_key, + worker_job_id=job.id, + ingestion_job_id=None, + mode="batch", + ) + if race is not None: + db.delete(job) + db.flush() + return race return EnqueuedResponse(worker_job_id=worker_job_id, status="pending", mode="batch") @@ -502,7 +641,9 @@ async def push_ingestion_lines(request: Request) -> PushLinesResponse: rejected = _reject_if_queue_full(db) if rejected is not None: return rejected # type: ignore[return-value] - job, stats = ingest_push_lines(db, raw_lines) + job, stats = ingest_push_lines( + db, raw_lines, scope=_scope_from_request(request) + ) return PushLinesResponse( ingestion_job_id=str(job.id), diff --git a/src/clients/v1.py b/src/clients/v1.py index 6209171..bd6cdaf 100644 --- a/src/clients/v1.py +++ b/src/clients/v1.py @@ -68,11 +68,15 @@ def _request( method: str, path: str, json_body: Optional[dict[str, Any]] = None, + extra_headers: Optional[dict[str, str]] = None, ) -> Any: + headers = self._headers() + if extra_headers: + headers.update(extra_headers) response = self._client.request( method, self._url(path), - headers=self._headers(), + headers=headers, json=json_body, ) if response.status_code >= 400: @@ -99,13 +103,26 @@ def get_config(self) -> dict[str, Any]: """GET /v1/config.""" return self._get("/v1/config") - def create_ingestion(self, **body: Any) -> dict[str, Any]: + def create_ingestion( + self, + *, + idempotency_key: Optional[str] = None, + **body: Any, + ) -> dict[str, Any]: """POST /v1/ingestions — enqueue an ingest job. Pass ``callback_url`` (http/https) for an HMAC-signed completion POST - when the batch worker reaches a terminal state. + when the batch worker reaches a terminal state. Pass + ``idempotency_key`` to send an ``Idempotency-Key`` header so a retry + within the TTL returns the original job. """ - return self._post("/v1/ingestions", body) + extra = {"Idempotency-Key": idempotency_key} if idempotency_key else None + return self._request( + "POST", + "/v1/ingestions", + json_body=_omit_none(body), + extra_headers=extra, + ) def list_ingestions(self) -> dict[str, Any]: """GET /v1/ingestions.""" diff --git a/src/config/settings.py b/src/config/settings.py index e80267c..ad365a0 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -63,6 +63,9 @@ class Settings(BaseSettings): webhook_max_retries: int = 5 webhook_timeout: float = 10.0 + # Idempotency-Key TTL for POST /v1/ingestions (batch enqueue and tail create). + ingest_idempotency_ttl_seconds: int = 86400 + # Adapters adapter_cloudwatch_region: str = "us-east-1" datadog_api_key: str = "" diff --git a/src/core/ingestion/idempotency.py b/src/core/ingestion/idempotency.py new file mode 100644 index 0000000..ee650e0 --- /dev/null +++ b/src/core/ingestion/idempotency.py @@ -0,0 +1,110 @@ +"""Request-level ingest idempotency (Idempotency-Key on POST /v1/ingestions).""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +import structlog +from sqlalchemy.exc import IntegrityError + +from src.db.models import IngestIdempotencyKey + +log = structlog.get_logger() + +MAX_IDEMPOTENCY_KEY_LENGTH = 256 + + +class InvalidIdempotencyKey(ValueError): + """Empty or oversized Idempotency-Key.""" + + +def parse_idempotency_key(raw: Optional[str]) -> Optional[str]: + """Return a stored key, or None when the header is omitted. + + Raises ``InvalidIdempotencyKey`` for blank or oversized values. + """ + if raw is None: + return None + key = raw.strip() + if not key: + raise InvalidIdempotencyKey("Idempotency-Key must be a non-empty string") + if len(key) > MAX_IDEMPOTENCY_KEY_LENGTH: + raise InvalidIdempotencyKey("Idempotency-Key exceeds 256 characters") + return key + + +def key_log_prefix(key: str) -> str: + """Short prefix safe to log — never the full key.""" + if len(key) <= 8: + return key[:2] + "…" if len(key) > 2 else "…" + return key[:8] + + +def get_idempotency_row(db: Any, key: str) -> Optional[IngestIdempotencyKey]: + return ( + db.query(IngestIdempotencyKey).filter(IngestIdempotencyKey.key == key).first() + ) + + +def is_active(row: IngestIdempotencyKey, now: datetime) -> bool: + expires = row.expires_at + if expires is None: + return False + if expires.tzinfo is None: + expires = expires.replace(tzinfo=timezone.utc) + compare = now if now.tzinfo is not None else now.replace(tzinfo=timezone.utc) + return expires > compare + + +def lookup_active_key( + db: Any, key: str, now: datetime +) -> Optional[IngestIdempotencyKey]: + row = get_idempotency_row(db, key) + if row is None or not is_active(row, now): + return None + return row + + +def store_idempotency_key( + db: Any, + key: str, + *, + worker_job_id: Optional[uuid.UUID], + ingestion_job_id: Optional[uuid.UUID], + mode: str, + now: datetime, + ttl_seconds: int, +) -> tuple[IngestIdempotencyKey, bool]: + """Insert key → job mapping. Returns ``(row, is_replay)``. + + Expired rows are deleted then replaced. A unique-key race returns the + winner's row as a replay. + """ + existing = get_idempotency_row(db, key) + if existing is not None: + if is_active(existing, now): + return existing, True + db.delete(existing) + db.flush() + + row = IngestIdempotencyKey( + key=key, + worker_job_id=worker_job_id, + ingestion_job_id=ingestion_job_id, + mode=mode, + created_at=now, + expires_at=now + timedelta(seconds=ttl_seconds), + ) + try: + with db.begin_nested(): + db.add(row) + db.flush() + return row, False + except IntegrityError: + winner = get_idempotency_row(db, key) + if winner is not None and is_active(winner, now): + log.info("ingest_idempotency_race", key_prefix=key_log_prefix(key)) + return winner, True + raise diff --git a/src/core/ingestion/service.py b/src/core/ingestion/service.py index a01f7d8..a1d1fb2 100644 --- a/src/core/ingestion/service.py +++ b/src/core/ingestion/service.py @@ -2,9 +2,11 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Callable, Iterable, Optional +from typing import Any, Callable, Iterable, Optional +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session +from sqlalchemy.sql.dml import Insert from src.adapters.base import SourceSpec, TimeWindow from src.adapters.file.adapter import detect_format, discover_files, read_lines @@ -14,7 +16,14 @@ from src.core.normalization.fingerprint import fingerprint_message from src.core.parsing.json_parser import ParsedLogLine, parse_json_line from src.core.parsing.text_parser import parse_text_line -from src.db.models import IngestionJob, LogEntry, Source +from src.db.models import ( + DEFAULT_LOG_SCOPE, + LOG_ENTRY_DEDUP_INDEX_WHERE, + IngestionJob, + LogEntry, + Source, +) +from src.utils.hashing import hash_raw_line @dataclass @@ -23,6 +32,7 @@ class IngestionStats: lines_read: int = 0 parsed_count: int = 0 error_count: int = 0 + deduped_count: int = 0 services_detected: set[str] = field(default_factory=set) duration_seconds: float = 0.0 @@ -59,11 +69,12 @@ def _process_line( source: Source, job: IngestionJob, source_adapter: str, - source_ref: str, + source_ref: Optional[str], stats: IngestionStats, received_at: Optional[datetime] = None, default_host: Optional[str] = None, extra: Optional[dict] = None, + scope: str = DEFAULT_LOG_SCOPE, ) -> Optional[LogEntry]: """Parse, fingerprint, and build a LogEntry for one raw line. Returns None for blank/unparseable/error lines (stats are updated either way). @@ -117,7 +128,9 @@ def _process_line( parser_type=parsed.parser_type, extra_json={**(extra or {}), **(parsed.extra or {})} or None, source_adapter=source_adapter, - source_ref=source_ref, + source_ref=source_ref or "", + original_line_hash=hash_raw_line(line), + scope=scope or DEFAULT_LOG_SCOPE, ) stats.parsed_count += 1 @@ -127,19 +140,92 @@ def _process_line( return entry +def _log_entry_values(entry: LogEntry) -> dict[str, Any]: + """Column values for a PostgreSQL INSERT of one log row.""" + return { + "id": entry.id, + "source_id": entry.source_id, + "ingestion_job_id": entry.ingestion_job_id, + "timestamp": entry.timestamp, + "service": entry.service, + "environment": entry.environment, + "level": entry.level, + "trace_id": entry.trace_id, + "request_id": entry.request_id, + "host": entry.host, + "raw_message": entry.raw_message, + "normalized_message": entry.normalized_message, + "fingerprint": entry.fingerprint, + "parser_type": entry.parser_type, + "extra_json": entry.extra_json, + "source_adapter": entry.source_adapter, + "source_ref": entry.source_ref if entry.source_ref is not None else "", + "original_line_hash": entry.original_line_hash, + "scope": entry.scope or DEFAULT_LOG_SCOPE, + } + + +def log_entry_upsert_statement(batch: list[LogEntry]) -> Insert: + """INSERT … ON CONFLICT DO NOTHING against ``ux_log_entries_dedup``.""" + stmt = pg_insert(LogEntry).values([_log_entry_values(entry) for entry in batch]) + return stmt.on_conflict_do_nothing( + index_elements=["scope", "source_ref", "original_line_hash", "timestamp"], + index_where=LOG_ENTRY_DEDUP_INDEX_WHERE, + ).returning(LogEntry.id) + + +def _entries_inserted(batch: list[LogEntry], result: Any) -> list[LogEntry]: + """Rows actually inserted. Mocks without RETURNING rows assume the full batch.""" + if result is None: + return list(batch) + fetchall = getattr(result, "fetchall", None) + if not callable(fetchall): + return list(batch) + try: + rows = fetchall() + except Exception: + return list(batch) + if not isinstance(rows, (list, tuple)): + return list(batch) + if not rows: + return [] + try: + inserted_ids = {row[0] for row in rows} + except (TypeError, IndexError): + return list(batch) + return [entry for entry in batch if entry.id in inserted_ids] + + def _flush_log_batch( db: Session, batch: list[LogEntry], embedder: Optional[EmbeddingsProvider], -) -> None: - """Persist a batch of log entries, then optionally their embeddings.""" +) -> tuple[int, int]: + """Persist a batch with content-dedup upsert. + + Returns ``(inserted_count, skipped_count)``. Duplicate physical lines + matching ``ux_log_entries_dedup`` are skipped, not errors. + """ if not batch: - return - db.bulk_save_objects(batch) - db.flush() - if embedder is not None: - persist_log_embeddings(db, batch, provider=embedder) + return 0, 0 + stmt = log_entry_upsert_statement(batch) + result = db.execute(stmt) + inserted_entries = _entries_inserted(batch, result) + skipped = len(batch) - len(inserted_entries) + if embedder is not None and inserted_entries: + persist_log_embeddings(db, inserted_entries, provider=embedder) batch.clear() + return len(inserted_entries), skipped + + +def _flush_and_count( + db: Session, + batch: list[LogEntry], + embedder: Optional[EmbeddingsProvider], + stats: IngestionStats, +) -> None: + _inserted, skipped = _flush_log_batch(db, batch, embedder) + stats.deduped_count += skipped def ingest_files( @@ -152,6 +238,7 @@ def ingest_files( fmt: str = "auto", progress_callback: Optional[Callable[[int, int], None]] = None, with_embeddings: bool = False, + scope: str = DEFAULT_LOG_SCOPE, ) -> tuple[IngestionJob, IngestionStats]: """ Main ingestion entry point for local files. Discovers, parses, and persists log files. @@ -220,6 +307,7 @@ def ingest_files( "file", str(file_path), stats, + scope=scope, ) if entry is None: continue @@ -227,13 +315,13 @@ def ingest_files( batch.append(entry) if len(batch) >= BATCH_SIZE: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) if progress_callback: progress_callback(stats.lines_read, stats.parsed_count) if batch: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) stats.duration_seconds = time.time() - start_time @@ -268,6 +356,7 @@ def ingest_from_source( with_embeddings: bool = False, existing_job: Optional[IngestionJob] = None, finalize: bool = True, + scope: str = DEFAULT_LOG_SCOPE, ) -> tuple[IngestionJob, IngestionStats]: """ Adapter-driven ingestion entry point (e.g. CloudWatch, Loki). Discovers streams via the @@ -391,6 +480,7 @@ def ingest_from_source( received_at=raw.received_at, default_host=raw.default_host, extra=raw.extra or None, + scope=scope, ) if entry is None: continue @@ -398,7 +488,7 @@ def ingest_from_source( batch.append(entry) if len(batch) >= BATCH_SIZE: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) if progress_callback: progress_callback(stats.lines_read, stats.parsed_count) @@ -417,7 +507,7 @@ def ingest_from_source( raise AdapterUnavailableError("; ".join(adapter_errors)) if batch: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) stats.duration_seconds = time.time() - start_time _apply_ingest_counts(job, stats, additive=existing_job is not None) @@ -459,6 +549,7 @@ def ingest_push_lines( default_env: Optional[str] = None, fmt: str = "auto", with_embeddings: bool = False, + scope: str = DEFAULT_LOG_SCOPE, ) -> tuple[IngestionJob, IngestionStats]: """Persist caller-pushed raw lines through parse → fingerprint → LogEntry.""" import time @@ -496,15 +587,16 @@ def ingest_push_lines( "push", "push", stats, + scope=scope, ) if entry is None: continue batch.append(entry) if len(batch) >= BATCH_SIZE: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) if batch: - _flush_log_batch(db, batch, embedder) + _flush_and_count(db, batch, embedder, stats) stats.duration_seconds = time.time() - start_time job.status = "completed" diff --git a/src/core/ingestion/tail.py b/src/core/ingestion/tail.py index 0940bab..3a7ece4 100644 --- a/src/core/ingestion/tail.py +++ b/src/core/ingestion/tail.py @@ -238,6 +238,7 @@ def tick_one_tail_job( with_embeddings=with_embeddings, existing_job=job, finalize=False, + scope=str(meta.get("scope") or "default"), ) except Exception as exc: new_count, should_pause = consecutive_errors_after_failure( diff --git a/src/db/__init__.py b/src/db/__init__.py index 3ba8dc7..dfee273 100644 --- a/src/db/__init__.py +++ b/src/db/__init__.py @@ -6,10 +6,12 @@ ClusterMember, ClusterRun, Explanation, + IngestIdempotencyKey, IngestionJob, LogEmbedding, LogEntry, Source, + WorkerJob, ) from .session import check_connection, get_db, get_engine @@ -25,6 +27,8 @@ "Explanation", "ApiKey", "AppConfig", + "IngestIdempotencyKey", + "WorkerJob", "get_db", "get_engine", "check_connection", diff --git a/src/db/models.py b/src/db/models.py index 92d047c..2a2e1d1 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -4,8 +4,6 @@ from pgvector.sqlalchemy import Vector from sqlalchemy import ( - BigInteger, - Boolean, DateTime, Float, ForeignKey, @@ -13,12 +11,18 @@ Integer, String, Text, - UniqueConstraint, + text, ) from sqlalchemy.dialects.postgresql import JSONB, UUID from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship from sqlalchemy.sql import func +DEFAULT_LOG_SCOPE = "default" +LOG_ENTRY_DEDUP_INDEX = "ux_log_entries_dedup" +LOG_ENTRY_DEDUP_INDEX_WHERE = text( + "original_line_hash IS NOT NULL AND timestamp IS NOT NULL AND source_ref IS NOT NULL" +) + class Base(DeclarativeBase): pass @@ -83,6 +87,10 @@ class LogEntry(Base): extra_json: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) source_adapter: Mapped[str] = mapped_column(String(50), nullable=False, default="file") source_ref: Mapped[str | None] = mapped_column(Text, nullable=True) + original_line_hash: Mapped[str | None] = mapped_column(String(64), nullable=True) + scope: Mapped[str] = mapped_column( + String(255), nullable=False, default=DEFAULT_LOG_SCOPE, server_default="default" + ) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) source: Mapped["Source"] = relationship("Source", back_populates="log_entries") @@ -93,6 +101,15 @@ class LogEntry(Base): __table_args__ = ( Index("ix_log_entries_timestamp_service", "timestamp", "service"), Index("ix_log_entries_source_adapter", "source_adapter"), + Index( + LOG_ENTRY_DEDUP_INDEX, + "scope", + "source_ref", + "original_line_hash", + "timestamp", + unique=True, + postgresql_where=LOG_ENTRY_DEDUP_INDEX_WHERE, + ), ) @@ -211,3 +228,20 @@ class WorkerJob(Base): created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class IngestIdempotencyKey(Base): + """Maps an Idempotency-Key header to the original ingest enqueue result.""" + + __tablename__ = "ingest_idempotency_keys" + + key: Mapped[str] = mapped_column(String(256), primary_key=True) + worker_job_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), ForeignKey("worker_jobs.id"), nullable=True + ) + ingestion_job_id: Mapped[uuid.UUID | None] = mapped_column( + UUID(as_uuid=True), ForeignKey("ingestion_jobs.id"), nullable=True + ) + mode: Mapped[str] = mapped_column(String(20), nullable=False, default="batch") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) diff --git a/src/utils/__init__.py b/src/utils/__init__.py index ceaf8e2..ff675e8 100644 --- a/src/utils/__init__.py +++ b/src/utils/__init__.py @@ -1,5 +1,5 @@ from .time import format_window, parse_duration, resolve_baseline_window, resolve_window -from .hashing import hash_dict, short_hash +from .hashing import hash_dict, hash_raw_line, short_hash __all__ = [ "parse_duration", @@ -7,5 +7,6 @@ "resolve_baseline_window", "format_window", "hash_dict", + "hash_raw_line", "short_hash", ] diff --git a/src/utils/hashing.py b/src/utils/hashing.py index 24285e4..9c6a2de 100644 --- a/src/utils/hashing.py +++ b/src/utils/hashing.py @@ -11,3 +11,8 @@ def hash_dict(data: dict[str, Any]) -> str: def short_hash(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest()[:16] + + +def hash_raw_line(text: str) -> str: + """SHA-256 hex digest of a raw log line (content-dedup key, not fingerprint).""" + return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() diff --git a/src/worker/runner.py b/src/worker/runner.py index 986e12d..1c3fe97 100644 --- a/src/worker/runner.py +++ b/src/worker/runner.py @@ -5,9 +5,9 @@ No external queue dependency — uses PostgreSQL as the queue. Safe to run multiple workers (uses SELECT FOR UPDATE SKIP LOCKED). """ + import signal import time -import uuid from datetime import datetime, timezone import structlog @@ -15,12 +15,12 @@ log = structlog.get_logger() POLL_INTERVAL = 2 # seconds between polls when idle -BATCH_SIZE = 1 # one job at a time per worker process +BATCH_SIZE = 1 # one job at a time per worker process def claim_next_job(db): """Atomically claim one pending worker job. Returns job or None.""" - from sqlalchemy import select, update + from sqlalchemy import select from src.db.models import WorkerJob # SELECT FOR UPDATE SKIP LOCKED ensures multiple workers don't double-claim @@ -58,6 +58,7 @@ def run_ingest_job(db, worker_job) -> dict: default_env=payload.get("env"), fmt=payload.get("format", "auto"), with_embeddings=payload.get("with_embeddings", False), + scope=payload.get("scope") or "default", ) else: import uuid @@ -70,23 +71,41 @@ def run_ingest_job(db, worker_job) -> dict: window = None if payload.get("since") or payload.get("from_time") or payload.get("to_time"): - from_dt = datetime.fromisoformat(payload["from_time"]) if payload.get("from_time") else None - to_dt = datetime.fromisoformat(payload["to_time"]) if payload.get("to_time") else None + from_dt = ( + datetime.fromisoformat(payload["from_time"]) + if payload.get("from_time") + else None + ) + to_dt = ( + datetime.fromisoformat(payload["to_time"]) + if payload.get("to_time") + else None + ) # resolve_window attaches UTC to naive datetimes and handles a single bound # (from-only -> "from X to now") — same helper /query/explain uses. - w_start, w_end = resolve_window(since=payload.get("since"), from_time=from_dt, to_time=to_dt) + w_start, w_end = resolve_window( + since=payload.get("since"), from_time=from_dt, to_time=to_dt + ) window = TimeWindow(start=w_start, end=w_end) resume_cursors = None resume_completed_streams = None if payload.get("resume_ingestion_job_id"): - prior = db.query(IngestionJob).filter( - IngestionJob.id == uuid.UUID(payload["resume_ingestion_job_id"]) - ).first() + prior = ( + db.query(IngestionJob) + .filter( + IngestionJob.id == uuid.UUID(payload["resume_ingestion_job_id"]) + ) + .first() + ) if prior is None: - raise ValueError(f"No ingestion job found with id {payload['resume_ingestion_job_id']}") + raise ValueError( + f"No ingestion job found with id {payload['resume_ingestion_job_id']}" + ) resume_cursors = (prior.metadata_json or {}).get("cursors") - resume_completed_streams = (prior.metadata_json or {}).get("completed_streams") + resume_completed_streams = (prior.metadata_json or {}).get( + "completed_streams" + ) params = payload.get("params", {}) if adapter in ("k8s", "kubernetes"): @@ -113,6 +132,7 @@ def run_ingest_job(db, worker_job) -> dict: resume_cursors=resume_cursors, resume_completed_streams=resume_completed_streams, with_embeddings=payload.get("with_embeddings", False), + scope=payload.get("scope") or "default", ) # Link worker job → ingestion job @@ -124,6 +144,7 @@ def run_ingest_job(db, worker_job) -> dict: "lines_read": stats.lines_read, "parsed_count": stats.parsed_count, "error_count": stats.error_count, + "deduped_count": stats.deduped_count, "services_detected": list(stats.services_detected), "duration_seconds": stats.duration_seconds, } @@ -131,8 +152,6 @@ def run_ingest_job(db, worker_job) -> dict: def process_one(db) -> bool: """Claim and execute one job. Returns True if a job was processed.""" - from src.db.models import WorkerJob - worker_job = claim_next_job(db) if worker_job is None: return False diff --git a/tests/integration/test_ingest_dedup.py b/tests/integration/test_ingest_dedup.py new file mode 100644 index 0000000..74d6c38 --- /dev/null +++ b/tests/integration/test_ingest_dedup.py @@ -0,0 +1,52 @@ +"""Content-dedup integration test. Skipped without Postgres (CI has no DB).""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.skipif( + not os.getenv("DB_URL") and not os.getenv("INTEGRATION_TESTS"), + reason="Integration tests require DB_URL environment variable", +) + + +@pytest.fixture +def db_session(): + from src.db.models import Base + from src.db.session import check_connection, get_db, get_engine + + if not check_connection(): + pytest.skip("Cannot connect to database") + + engine = get_engine() + from sqlalchemy import text + + with engine.connect() as conn: + conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + conn.commit() + Base.metadata.create_all(engine) + + with get_db() as db: + yield db + + +def test_reingest_identical_lines_keeps_single_row(db_session, tmp_path: Path) -> None: + from sqlalchemy import func, select + + from src.core.ingestion.service import ingest_files + from src.db.models import LogEntry + + log_file = tmp_path / "dup.log" + log_file.write_text( + '{"timestamp": "2026-01-01T00:00:00Z", "message": "boom", "level": "error", "service": "api"}\n' + ) + + ingest_files(db=db_session, paths=[str(log_file)]) + ingest_files(db=db_session, paths=[str(log_file)]) + db_session.flush() + + count = db_session.execute(select(func.count()).select_from(LogEntry)).scalar_one() + assert count == 1 diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index 3728020..6b6f64a 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -389,6 +389,7 @@ def capture_add(obj): assert resp.status_code == 202 assert "callback_url" not in added[0].payload_json + assert added[0].payload_json["scope"] == "default" def test_422_on_file_callback_url(self): with patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]): @@ -445,6 +446,45 @@ def capture_add(obj): assert added[0].payload_json["api_key_id"] == str(key_id) assert added[0].payload_json["scope"] == "incident:INC-9" + def test_scope_on_payload_without_callback_url(self): + from src.config.settings import Settings + + key_id = uuid.uuid4() + record = MagicMock() + record.id = key_id + record.name = "ingest" + record.key_prefix = "rlk_testhash" + record.role = "ingest" + record.scope = "incident:A" + record.revoked_at = None + + mock_db = _ctx_db() + added = [] + + def capture_add(obj): + added.append(obj) + obj.id = uuid.uuid4() + + mock_db.add.side_effect = capture_add + settings = Settings(_env_file=None, auth_enabled=True, auth_mode="api_key") + + with patch("src.config.get_settings", return_value=settings), \ + patch("src.api.auth.keys.lookup_api_key", return_value=record), \ + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), \ + patch("src.db.session.get_db", side_effect=lambda: mock_db): + resp = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Authorization": "Bearer rlk_ingestrole1"}, + ) + + assert resp.status_code == 202 + jobs = [obj for obj in added if hasattr(obj, "payload_json")] + assert jobs, "expected a WorkerJob to be enqueued" + assert "callback_url" not in jobs[0].payload_json + assert jobs[0].payload_json["scope"] == "incident:A" + assert jobs[0].payload_json["api_key_id"] == str(key_id) + # ── GET /ingestions/jobs/{id} ───────────────────────────────────────────────── diff --git a/tests/unit/test_ingest_dedup.py b/tests/unit/test_ingest_dedup.py new file mode 100644 index 0000000..332d905 --- /dev/null +++ b/tests/unit/test_ingest_dedup.py @@ -0,0 +1,154 @@ +"""Unit tests for raw-line hashing and content-dedup upsert.""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from unittest.mock import MagicMock + +from sqlalchemy.dialects import postgresql + +from src.core.ingestion.service import ( + IngestionStats, + _flush_log_batch, + _process_line, + log_entry_upsert_statement, +) +from src.core.normalization.fingerprint import fingerprint_message +from src.db.models import LogEntry +from src.utils.hashing import hash_raw_line + + +def test_hash_raw_line_is_stable_and_distinct_from_fingerprint() -> None: + line = '{"message": "timeout talking to payments id=abc-123", "level": "error"}' + first = hash_raw_line(line) + second = hash_raw_line(line) + assert first == second + assert len(first) == 64 + assert first != hash_raw_line(line + " ") + _normalized, fingerprint = fingerprint_message( + "timeout talking to payments id=abc-123" + ) + assert first != fingerprint + + +def test_process_line_sets_hash_scope_and_empty_source_ref() -> None: + source = MagicMock() + source.id = uuid.uuid4() + job = MagicMock() + job.id = uuid.uuid4() + stats = IngestionStats() + line = '{"message": "hello 99", "level": "info", "service": "api"}' + entry = _process_line( + line, + "json", + None, + None, + source, + job, + "file", + None, + stats, + ) + assert entry is not None + assert entry.original_line_hash == hash_raw_line(line) + assert entry.scope == "default" + assert entry.source_ref == "" + assert entry.fingerprint != entry.original_line_hash + assert stats.parsed_count == 1 + + +def test_process_line_stamps_caller_scope() -> None: + source = MagicMock() + source.id = uuid.uuid4() + job = MagicMock() + job.id = uuid.uuid4() + entry = _process_line( + "plain info line", + "text", + "svc", + None, + source, + job, + "push", + "push", + IngestionStats(), + scope="incident:INC-9", + ) + assert entry is not None + assert entry.scope == "incident:INC-9" + assert entry.source_ref == "push" + assert entry.original_line_hash == hash_raw_line("plain info line") + + +def test_upsert_statement_compiles_on_conflict_do_nothing() -> None: + entry = LogEntry( + id=uuid.uuid4(), + source_adapter="file", + source_ref="app.log", + original_line_hash="a" * 64, + scope="default", + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + fingerprint="abcd1234abcd1234", + raw_message="hello", + ) + stmt = log_entry_upsert_statement([entry]) + compiled = str(stmt.compile(dialect=postgresql.dialect())).upper() + assert "ON CONFLICT" in compiled + assert "DO NOTHING" in compiled + assert "SCOPE" in compiled + assert "SOURCE_REF" in compiled + assert "ORIGINAL_LINE_HASH" in compiled + assert "TIMESTAMP" in compiled + assert "WHERE" in compiled + assert "ORIGINAL_LINE_HASH IS NOT NULL" in compiled + assert "TIMESTAMP IS NOT NULL" in compiled + assert "SOURCE_REF IS NOT NULL" in compiled + + +def test_flush_log_batch_executes_on_conflict_insert() -> None: + entry = LogEntry( + id=uuid.uuid4(), + source_adapter="file", + source_ref="app.log", + original_line_hash="b" * 64, + scope="default", + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + fingerprint="ffff", + ) + db = MagicMock() + result = MagicMock() + result.fetchall.return_value = [(entry.id,)] + db.execute.return_value = result + + inserted, skipped = _flush_log_batch(db, [entry], None) + + assert inserted == 1 + assert skipped == 0 + db.execute.assert_called_once() + stmt = db.execute.call_args[0][0] + compiled = str(stmt.compile(dialect=postgresql.dialect())).upper() + assert "ON CONFLICT" in compiled + assert "DO NOTHING" in compiled + + +def test_flush_log_batch_skips_duplicates_when_returning_empty() -> None: + entry = LogEntry( + id=uuid.uuid4(), + source_adapter="file", + source_ref="app.log", + original_line_hash="c" * 64, + scope="default", + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + db = MagicMock() + result = MagicMock() + result.fetchall.return_value = [] + db.execute.return_value = result + embedder = MagicMock() + + inserted, skipped = _flush_log_batch(db, [entry], embedder) + + assert inserted == 0 + assert skipped == 1 + embedder.embed_texts.assert_not_called() diff --git a/tests/unit/test_ingest_idempotency.py b/tests/unit/test_ingest_idempotency.py new file mode 100644 index 0000000..496eb51 --- /dev/null +++ b/tests/unit/test_ingest_idempotency.py @@ -0,0 +1,298 @@ +"""Unit tests for Idempotency-Key on POST /v1/ingestions.""" + +from __future__ import annotations + +import uuid +from contextlib import nullcontext +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.exc import IntegrityError + +from src.api.app import app +from src.core.ingestion.idempotency import ( + InvalidIdempotencyKey, + is_active, + key_log_prefix, + parse_idempotency_key, + store_idempotency_key, +) +from src.db.models import IngestIdempotencyKey, WorkerJob + +client = TestClient(app, raise_server_exceptions=False) + + +def _ctx_db() -> MagicMock: + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.filter.return_value.first.return_value = None + mock_db.execute.return_value.scalar_one.return_value = 0 + mock_db.begin_nested.return_value = nullcontext() + return mock_db + + +class TestParseIdempotencyKey: + def test_none_is_omitted(self) -> None: + assert parse_idempotency_key(None) is None + + def test_strips_and_keeps_value(self) -> None: + assert parse_idempotency_key(" abc-123 ") == "abc-123" + + def test_empty_raises(self) -> None: + with pytest.raises(InvalidIdempotencyKey, match="non-empty"): + parse_idempotency_key("") + with pytest.raises(InvalidIdempotencyKey, match="non-empty"): + parse_idempotency_key(" ") + + def test_too_long_raises(self) -> None: + with pytest.raises(InvalidIdempotencyKey, match="256"): + parse_idempotency_key("k" * 257) + + def test_log_prefix_is_short(self) -> None: + prefix = key_log_prefix("super-secret-idempotency-key") + assert prefix == "super-se" + assert "secret-idempotency" not in prefix + + +class TestIsActive: + def test_future_expiry_is_active(self) -> None: + row = MagicMock() + row.expires_at = datetime(2026, 8, 18, tzinfo=timezone.utc) + now = datetime(2026, 8, 17, tzinfo=timezone.utc) + assert is_active(row, now) is True + + def test_past_expiry_is_inactive(self) -> None: + row = MagicMock() + row.expires_at = datetime(2026, 8, 16, tzinfo=timezone.utc) + now = datetime(2026, 8, 17, tzinfo=timezone.utc) + assert is_active(row, now) is False + + +class TestStoreIdempotencyKey: + def test_unique_conflict_returns_existing(self) -> None: + winner_id = uuid.uuid4() + winner = IngestIdempotencyKey( + key="same", + worker_job_id=winner_id, + ingestion_job_id=None, + mode="batch", + created_at=datetime.now(tz=timezone.utc), + expires_at=datetime.now(tz=timezone.utc) + timedelta(hours=1), + ) + db = MagicMock() + db.query.return_value.filter.return_value.first.side_effect = [None, winner] + db.begin_nested.return_value = nullcontext() + db.flush.side_effect = IntegrityError("INSERT", {}, Exception("unique")) + + row, replay = store_idempotency_key( + db, + "same", + worker_job_id=uuid.uuid4(), + ingestion_job_id=None, + mode="batch", + now=datetime.now(tz=timezone.utc), + ttl_seconds=86400, + ) + assert replay is True + assert row.worker_job_id == winner_id + + +class TestCreateIngestionIdempotency: + def test_same_key_returns_original_worker_job_id(self) -> None: + stored: dict[str, IngestIdempotencyKey | None] = {"row": None} + mock_db = _ctx_db() + + def query_side_effect(model: object) -> MagicMock: + q = MagicMock() + name = getattr(model, "__name__", "") + if name == "IngestIdempotencyKey": + q.filter.return_value.first.return_value = stored["row"] + else: + q.filter.return_value.first.return_value = None + return q + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = uuid.uuid4() + if isinstance(obj, IngestIdempotencyKey): + stored["row"] = obj + if getattr(obj, "expires_at", None) is None: + obj.expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=1) + + mock_db.query.side_effect = query_side_effect + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + first = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "req-1"}, + ) + second = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"idempotency-key": "req-1"}, + ) + + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["worker_job_id"] == second.json()["worker_job_id"] + + def test_same_key_on_unversioned_alias(self) -> None: + stored: dict[str, IngestIdempotencyKey | None] = {"row": None} + mock_db = _ctx_db() + + def query_side_effect(model: object) -> MagicMock: + q = MagicMock() + name = getattr(model, "__name__", "") + if name == "IngestIdempotencyKey": + q.filter.return_value.first.return_value = stored["row"] + else: + q.filter.return_value.first.return_value = None + return q + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = uuid.uuid4() + if isinstance(obj, IngestIdempotencyKey): + stored["row"] = obj + if getattr(obj, "expires_at", None) is None: + obj.expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=1) + + mock_db.query.side_effect = query_side_effect + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + first = client.post( + "/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "alias-1"}, + ) + second = client.post( + "/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "alias-1"}, + ) + + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["worker_job_id"] == second.json()["worker_job_id"] + + def test_different_keys_create_two_jobs(self) -> None: + ids: list[uuid.UUID] = [] + mock_db = _ctx_db() + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = uuid.uuid4() + ids.append(obj.id) + + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + first = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "key-a"}, + ) + second = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "key-b"}, + ) + + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["worker_job_id"] != second.json()["worker_job_id"] + assert len(ids) == 2 + + def test_expired_key_starts_a_new_job(self) -> None: + old_id = uuid.uuid4() + new_id = uuid.uuid4() + expired = MagicMock() + expired.key = "stale" + expired.worker_job_id = old_id + expired.ingestion_job_id = None + expired.mode = "batch" + expired.expires_at = datetime(2020, 1, 1, tzinfo=timezone.utc) + + mock_db = _ctx_db() + mock_db.query.return_value.filter.return_value.first.return_value = expired + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = new_id + + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + resp = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "stale"}, + ) + + assert resp.status_code == 202 + assert resp.json()["worker_job_id"] == str(new_id) + assert resp.json()["worker_job_id"] != str(old_id) + mock_db.delete.assert_called() + + def test_empty_key_is_400(self) -> None: + with patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]): + resp = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": " "}, + ) + assert resp.status_code == 400 + assert "non-empty" in resp.json()["detail"] + + def test_oversized_key_is_400(self) -> None: + resp = client.post( + "/v1/ingestions", + json={"paths": ["/logs"]}, + headers={"Idempotency-Key": "k" * 257}, + ) + assert resp.status_code == 400 + assert "256" in resp.json()["detail"] + + def test_omitted_key_keeps_current_behavior(self) -> None: + job_id = uuid.uuid4() + mock_db = _ctx_db() + + def capture_add(obj: object) -> None: + if isinstance(obj, WorkerJob): + obj.id = job_id + + mock_db.add.side_effect = capture_add + + with ( + patch("src.adapters.file.adapter.discover_files", return_value=["f.log"]), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + ): + resp = client.post("/v1/ingestions", json={"paths": ["/logs"]}) + + assert resp.status_code == 202 + assert resp.json()["worker_job_id"] == str(job_id) + added_keys = [ + obj + for call in mock_db.add.call_args_list + for obj in call.args + if isinstance(obj, IngestIdempotencyKey) + ] + assert added_keys == [] diff --git a/tests/unit/test_ingestion_service.py b/tests/unit/test_ingestion_service.py index 29a2428..e44b453 100644 --- a/tests/unit/test_ingestion_service.py +++ b/tests/unit/test_ingestion_service.py @@ -105,6 +105,21 @@ def _patch_get_adapter(monkeypatch, adapter): ) +def _capture_flushed_entries(monkeypatch: pytest.MonkeyPatch) -> list: + """Snapshot LogEntry batches passed to ``_flush_log_batch``.""" + from src.core.ingestion import service as svc + + captured: list = [] + original = svc._flush_log_batch + + def wrapped(db, batch, embedder): + captured.extend(list(batch)) + return original(db, batch, embedder) + + monkeypatch.setattr(svc, "_flush_log_batch", wrapped) + return captured + + class TestIngestFromSource: def test_happy_path(self, monkeypatch): db = _mock_db() @@ -131,6 +146,7 @@ def test_raw_line_defaults_fill_service_env_host(self, monkeypatch): fake.default_host = "billing-worker-abc" fake.extra = {"kubernetes": {"pod": "billing-worker-abc"}} _patch_get_adapter(monkeypatch, fake) + entries = _capture_flushed_entries(monkeypatch) job, stats = ingest_from_source( db=db, spec=SourceSpec(adapter="fake", params={}), window=_WINDOW @@ -138,9 +154,7 @@ def test_raw_line_defaults_fill_service_env_host(self, monkeypatch): assert job.status == "completed" assert stats.parsed_count == 1 - saved = [c for c in db.bulk_save_objects.call_args_list] - assert saved, "expected at least one bulk_save_objects call" - entries = saved[0].args[0] + assert entries, "expected at least one flushed log batch" assert entries[0].service == "billing-worker" assert entries[0].environment == "production" assert entries[0].host == "billing-worker-abc" @@ -227,8 +241,7 @@ def test_timestamp_falls_back_to_received_at(self, monkeypatch): query (clustering/explain filter on a non-null in-window timestamp). """ db = _mock_db() - entries = [] - db.bulk_save_objects.side_effect = lambda objs: entries.extend(objs) + entries = _capture_flushed_entries(monkeypatch) event_time = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) fake = FakeAdapter( @@ -244,7 +257,7 @@ def test_timestamp_falls_back_to_received_at(self, monkeypatch): assert len(entries) == 1 assert entries[0].timestamp == event_time - def test_raw_line_defaults_fill_service_env_host(self, monkeypatch): + def test_label_adapter_defaults_fill_service_env_host(self, monkeypatch): """Adapters (Loki labels, k8s fields) may set RawLogLine defaults; ingestion should persist them when the parsed line has no service/env/host.""" @@ -264,8 +277,7 @@ def read(self, ref, window): ) db = _mock_db() - entries = [] - db.bulk_save_objects.side_effect = lambda objs: entries.extend(objs) + entries = _capture_flushed_entries(monkeypatch) _patch_get_adapter(monkeypatch, LabelAdapter()) job, stats = ingest_from_source( @@ -297,8 +309,7 @@ def read(self, ref, window): ) db = _mock_db() - entries = [] - db.bulk_save_objects.side_effect = lambda objs: entries.extend(objs) + entries = _capture_flushed_entries(monkeypatch) _patch_get_adapter(monkeypatch, LabelAdapter()) spec = SourceSpec(adapter="labels", params={}, service="cli-svc", env="staging") @@ -494,7 +505,7 @@ def test_exception_mid_ingest_marks_job_failed(self, tmp_path): log_file.write_text('{"message": "hello", "level": "info"}\n') db = _mock_db() - db.bulk_save_objects.side_effect = RuntimeError("db exploded") + db.execute.side_effect = RuntimeError("db exploded") with pytest.raises(RuntimeError): ingest_files(db=db, paths=[str(tmp_path)]) diff --git a/tests/unit/test_openapi_contract.py b/tests/unit/test_openapi_contract.py index c456aab..d17e96f 100644 --- a/tests/unit/test_openapi_contract.py +++ b/tests/unit/test_openapi_contract.py @@ -68,6 +68,17 @@ def test_ingest_request_includes_callback_url() -> None: assert "callback_url" in schema["properties"] +def test_ingest_post_documents_idempotency_key_header() -> None: + paths = app.openapi()["paths"] + for path in ("/v1/ingestions", "/ingestions"): + params = paths[path]["post"].get("parameters") or [] + names = {p.get("name") for p in params} + assert "Idempotency-Key" in names, f"{path} POST must document Idempotency-Key" + header = next(p for p in params if p.get("name") == "Idempotency-Key") + assert header.get("in") == "header" + assert header.get("required") is not True + + def test_health_is_unversioned() -> None: paths = app.openapi()["paths"] assert "/health" in paths diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py index dc1ef39..b3a99de 100644 --- a/tests/unit/test_settings.py +++ b/tests/unit/test_settings.py @@ -120,3 +120,14 @@ def test_webhook_settings_from_env(monkeypatch): assert settings.webhook_secret == "whsec_global" assert settings.webhook_max_retries == 3 assert settings.webhook_timeout == pytest.approx(7.5) + + +def test_ingest_idempotency_ttl_default(): + settings = Settings(_env_file=None) + assert settings.ingest_idempotency_ttl_seconds == 86400 + + +def test_ingest_idempotency_ttl_from_env(monkeypatch): + monkeypatch.setenv("INGEST_IDEMPOTENCY_TTL_SECONDS", "60") + settings = Settings(_env_file=None) + assert settings.ingest_idempotency_ttl_seconds == 60