From d3104d66ab7b5571d69014e3b2d412980c2445ff Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 22 Sep 2026 16:51:51 -0300 Subject: [PATCH] fix: address review feedback on batch-size default, requirements-local, docstring - NOTIFICATION_QUEUE_BATCH_SIZE default reverted to 0 (unlimited) -- the existing behavior before this ticket, so deployments must opt in to the cap rather than having it silently applied at 5000 - requirements-local.txt: bump sientia_do pin to 1.12.3 to match requirements.txt - Trim the load_latest_data docstring addition so it reads as documentation, not a changelog entry Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 2 +- orchestrator/activities/mongo_db.py | 12 +++--------- orchestrator/utils/connectors_config.py | 2 +- requirements-local.txt | 2 +- tests/orchestrator/utils/test_connectors_config.py | 2 +- 5 files changed, 7 insertions(+), 13 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index c785e7b..3b029a0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -45,7 +45,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable ### Layering - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`). -- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default 5000), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run. +- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run. - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections. - **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing. diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index c3c8972..d101658 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -451,15 +451,9 @@ class MongoDB(SientiaMonitoring): filtering by timestamp to enable incremental data processing. It handles connection management and provides comprehensive error reporting. - SIENTIAPDE-2013: the query is capped at ``notification_queue_batch_size`` - documents (hard cap per run, no drain loop) and sorted by ``timestamp`` - ascending, so a capped run always returns the *oldest* pending documents - first. That matters because the caller advances its incremental cursor to - the max timestamp of whatever this returns -- without the ascending sort, - a capped batch could skip over older documents that never get picked up - on a later run. This backstops a cold start, backlog buildup, or - notification storm from fetching an unbounded (and potentially - multi-megabyte) payload in one query. + The query is capped at ``notification_queue_batch_size`` documents and sorted by + ``timestamp`` ascending, so a capped run returns the oldest pending documents first -- + the caller advances its cursor only to what was actually returned, so nothing is skipped. Args: input_data (dict[str, Any]): Activity input parameters. diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 14b2747..10e1769 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -33,7 +33,7 @@ def build_mongodb_config(): 'connection_string': connection_string, 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600, - 'notification_queue_batch_size': int(getenv('NOTIFICATION_QUEUE_BATCH_SIZE', '5000')), + 'notification_queue_batch_size': int(getenv('NOTIFICATION_QUEUE_BATCH_SIZE', '0')), } diff --git a/requirements-local.txt b/requirements-local.txt index 8df6bac..e25ae46 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.3 prometheus-client diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py index 8409d0b..fc8ad2a 100644 --- a/tests/orchestrator/utils/test_connectors_config.py +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -84,7 +84,7 @@ def test_build_mongo_db_config_with_defaults(): 'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018', 'database_name': 'sientia', 'ttl_index_seconds': 3600, - 'notification_queue_batch_size': 5000, + 'notification_queue_batch_size': 0, }