Limit stored responses; optimize status poll

config/config.default.yaml

@@ -44,6 +44,7 @@ job:
   ttl_max: 604800 # Time in seconds after which job state is cleared, regardless of last touch
 log:
   log_to_mongo: true # Persist logs in MongoDB.
+  mongo_level: INFO # Minimum level of logs persisted to MongoDB.
   mongo_ttl: 604800 # Time in seconds for a log to persist.
 redis:
   host: localhost
@@ -65,6 +66,7 @@ mongo:
   shutdown_timeout: 3 # Time in seconds to wait for serialize task to finish.
   compression_level: 10 # zstd level used to compress stored job documents. Higher means greater compression, at greater CPU cost.
   max_stored_doc_bytes: 134217728 # Compressed job-docs at or above this size are not persisted.
+  stored_success_proportion: 0.1 # Fraction of succeeded response bodies to persist. Failures are always kept.
 tier0:
   backend: gandalf
   backend_infores: infores:dogpark-tier0

pyproject.toml

@@ -139,6 +139,7 @@ launch = 'retriever'
     JOB__LOOKUP__TIER1_TIMEOUT=-1 \
     JOB__LOOKUP__TIER2_TIMEOUT=-1 \
     JOB__METAKG__TIMEOUT=-1 \
+    MONGO__STORED_SUCCESS_PROPORTION=1.0 \
     WORKERS=1 \
     retriever
 """

src/retriever/config/general.py

@@ -105,6 +105,14 @@ class MongoSettings(BaseModel):
             description="Compressed job-docs at or above this size are not persisted."
         ),
     ] = 128 * 1024**2
+    stored_success_proportion: Annotated[
+        float,
+        Field(
+            ge=0,
+            le=1,
+            description="Fraction of succeeded response bodies to persist. Failures are always kept.",
+        ),
+    ] = 0.1
 
 
 class TelemetrySettings(BaseModel):
@@ -231,6 +239,9 @@ class LogSettings(BaseModel):
     """Settings for log handling."""
 
     log_to_mongo: Annotated[bool, Field(description="Persist logs in MongoDB.")] = True
+    mongo_level: Annotated[
+        LogLevel, Field(description="Minimum level of logs persisted to MongoDB.")
+    ] = "INFO"
     mongo_ttl: Annotated[
         int, Field(description="Time in seconds for a log to persist.")
     ] = 604_800

src/retriever/query.py

@@ -260,14 +260,14 @@ async def in_progress_payload(
 ) -> tuple[HTTPStatus, dict[str, Any]]:
     """Return an AsyncQueryStatusResponse for in-progress or missing job.
 
-    "Running" if the row exists or there's log evidence, "Error" if the
-    fetch raised, otherwise "Not Found".
+    "Running" if the row exists, "Error" if the fetch raised, otherwise "Not
+    Found". Status polling returns empty log list for performance. Logs can be obtained
+    when query completes, or looking at /logs.
     """
-    logs = await job_logs(job_id)
-    if exists or len(logs) > 0:
+    if exists:
         return HTTPStatus.OK, {
             "status": "Running",
-            "logs": logs,
+            "logs": [],
             "description": "Job is running.",
         }
     if error is not None:
@@ -286,7 +286,11 @@ async def in_progress_payload(
 
 
 async def terminal_status_payload(
-    job_id: str, request: Request, status_doc: JobStatus
+    job_id: str,
+    request: Request,
+    status_doc: JobStatus,
+    *,
+    include_logs: bool = False,
 ) -> dict[str, Any]:
     """Build an AsyncQueryStatusResponse for a terminal job.
 
@@ -307,7 +311,7 @@ async def terminal_status_payload(
     return {
         "status": to_async_lifecycle(job_status),
         "description": description,
-        "logs": await job_logs(job_id),
+        "logs": await job_logs(job_id) if include_logs else [],
         "response_url": f"{request.base_url}response/{job_id}",
     }
 
@@ -362,7 +366,9 @@ async def get_job_response(
 
     # If job is abandoned, respond with the query
     if status_doc.get("abandoned"):
-        payload = await terminal_status_payload(job_id, request, status_doc)
+        payload = await terminal_status_payload(
+            job_id, request, status_doc, include_logs=True
+        )
         if job is not None and job.get("doc") is not None:
             query = unpack_doc(job)
             payload = {
@@ -372,7 +378,9 @@ async def get_job_response(
         return HTTPStatus.OK, payload
 
     if job is None or job.get("doc") is None:
-        return HTTPStatus.OK, await terminal_status_payload(job_id, request, status_doc)
+        return HTTPStatus.OK, await terminal_status_payload(
+            job_id, request, status_doc, include_logs=True
+        )
 
     response = unpack_doc(job)
     return HTTPStatus.OK, {

src/retriever/utils/logs.py

@@ -229,7 +229,7 @@ def mongo_sink(message: loguru.Message) -> None:
         serialize=True,
         enqueue=True,
         filter=lambda record: not record["extra"].get("no_mongo_log", False),
-        level=CONFIG.log_level,
+        level=CONFIG.log.mongo_level,
     )
 
 

src/retriever/utils/mongo.py

@@ -1623,7 +1623,7 @@ async def get_logs(
         """Return a generator of a filtered set of logs."""
         query = _build_log_query(start, end, level)
         if job_id is not None:
-            query["extra.job_id"] = {"$regex": job_id}
+            query["extra.job_id"] = job_id
 
         async for document in self._yield_logs(query):
             yield document
@@ -1747,6 +1747,16 @@ async def wrapup(self) -> None:
         self.client.close()
 
 
+def _store_job_success_response(job_id: str) -> bool:
+    """Whether to persist this succeeded response body.
+
+    Deterministic and uniform over `job_id` (a random uuid4 hex), so the two
+    writes an async job makes agree and never churn a stored blob.
+    `stored_success_proportion` of 0.0 keeps nothing, 1.0 keeps everything.
+    """
+    return int(job_id[:8], 16) / 0x1_0000_0000 < CONFIG.mongo.stored_success_proportion
+
+
 class MongoOutage(Exception):
     """Raised by `MongoQueue.put` when MongoClient is down; callers attach a warning."""
 
@@ -1811,6 +1821,13 @@ async def job_state(self, batch: list[QueryState | ResponseState]) -> None:
                     f"Job {state['job_id']} {doc_type} is {size} bytes (>= {CONFIG.mongo.max_stored_doc_bytes} cap); delivering without storing it.",
                     no_mongo_log=True,
                 )
+            elif (
+                doc_type == "response"
+                and cast("ResponseState", state)["status"] in TERMINAL_SUCCESS
+                and not _store_job_success_response(state["job_id"])
+            ):
+                # Don't store (but keep the query)
+                store_doc = False
             elif size > GRIDFS_INLINE_LIMIT:
                 doc_ref = await self.client.offload_doc_blob(state["job_id"], blob)  # pyright: ignore[reportArgumentType] size>0 implies blob is not None
             ops.append(

tests/test_query.py

@@ -5,6 +5,7 @@
 relevant query function.
 """
 
+from http import HTTPStatus
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -187,3 +188,97 @@ async def test_async_lookup_reapplies_data_tier_tag():
         call.args and call.args[0].get("data_tier") == 2
         for call in set_tags.call_args_list
     )
+
+
+def _status_request() -> SimpleNamespace:
+    """A stand-in Request exposing only the `base_url` these payloads read."""
+    return SimpleNamespace(base_url="http://test/")
+
+
+def _mongo_stub(
+    *,
+    status_doc: dict[str, object] | None = None,
+    job_doc: dict[str, object] | None = None,
+) -> Mock:
+    """A MongoClient() whose status/doc reads return the given fixtures."""
+    mongo = Mock()
+    mongo.get_job_status = AsyncMock(return_value=status_doc)
+    mongo.get_job_doc = AsyncMock(return_value=job_doc)
+    return mongo
+
+
+@pytest.mark.asyncio
+async def test_asyncquery_status_terminal_returns_empty_logs_without_scan():
+    """`/asyncquery_status` on a terminal job returns [] logs and never scans log_dump."""
+    mongo = _mongo_stub(status_doc={"status": "Success", "description": None})
+    with (
+        patch.object(query_module, "MongoClient", return_value=mongo),
+        patch.object(
+            query_module, "job_logs", AsyncMock(return_value=["scanned"])
+        ) as jl,
+    ):
+        code, payload = await query_module.get_job_status("abc123", _status_request())
+
+    assert code == HTTPStatus.OK
+    assert payload["logs"] == []
+    jl.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_status_in_progress_returns_empty_logs_without_scan():
+    """A Running job reports Running with [] logs and no log_dump scan."""
+    mongo = _mongo_stub(status_doc={"status": "Running", "description": None})
+    with (
+        patch.object(query_module, "MongoClient", return_value=mongo),
+        patch.object(
+            query_module, "job_logs", AsyncMock(return_value=["scanned"])
+        ) as jl,
+    ):
+        code, payload = await query_module.get_job_status("abc123", _status_request())
+
+    assert code == HTTPStatus.OK
+    assert payload["status"] == "Running"
+    assert payload["logs"] == []
+    jl.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_response_completed_stored_uses_blob_logs_without_scan():
+    """`/response` for a stored completed job serves the blob's logs, no scan."""
+    mongo = _mongo_stub(
+        status_doc={"status": "Success", "description": None},
+        job_doc={"job_id": "abc123", "doc": b"blob"},
+    )
+    blob = {"message": {}, "logs": ["from-blob"]}
+    with (
+        patch.object(query_module, "MongoClient", return_value=mongo),
+        patch.object(query_module, "unpack_doc", return_value=blob),
+        patch.object(
+            query_module, "job_logs", AsyncMock(return_value=["scanned"])
+        ) as jl,
+    ):
+        code, payload = await query_module.get_job_response("abc123", _status_request())
+
+    assert code == HTTPStatus.OK
+    assert payload["logs"] == ["from-blob"]
+    jl.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_response_completed_not_stored_falls_back_to_scan():
+    """`/response` for a completed-but-unstored job falls back to a log_dump scan."""
+    mongo = _mongo_stub(
+        status_doc={"status": "Success", "description": None},
+        job_doc=None,  # sampled out: no stored body
+    )
+    with (
+        patch.object(query_module, "MongoClient", return_value=mongo),
+        patch.object(
+            query_module, "job_logs", AsyncMock(return_value=["scanned"])
+        ) as jl,
+    ):
+        code, payload = await query_module.get_job_response("abc123", _status_request())
+
+    assert code == HTTPStatus.OK
+    assert payload["logs"] == ["scanned"]
+    jl.assert_awaited_once()