From 9c153041adc14f006caec615d26232ddfe4b7a1d Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Thu, 17 Sep 2026 09:33:46 -0300 Subject: [PATCH] SIENTIAPDE-2013: cap load_latest_data's Mongo fetch at a configurable batch size A cold start or notification storm had no backstop: load_latest_data fetched the entire matching backlog in one query. Confirmed live during SIENTIAPDE-2012's cluster verification -- an empty Redis cursor plus a real recurring error let a single run's query exceed Temporal's 4MB message limit and get stuck retrying forever. Threads a new notification_queue_batch_size config (env var NOTIFICATION_QUEUE_BATCH_SIZE, default 5000) through to the repository's find(), hard cap per run, no drain loop -- the incremental last_data_timestamp cursor picks up any remainder next run, per the parent ticket's design decision. Sorts by timestamp ascending so a capped run always returns the oldest pending documents first; without that, a capped batch has no guarantee of hitting the oldest documents, and the cursor would advance past whatever it skipped. Requires sientia_do>=1.13.0 (adds limit/sort params to MongoDBRepository.find in both the sync and async client) -- bumped the pin here per the parent ticket's decision to do it in the same PR. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 2 +- orchestrator/activities/activities.py | 1 + orchestrator/activities/mongo_db.py | 22 ++++++++- orchestrator/utils/connectors_config.py | 4 +- requirements.txt | 2 +- .../activities/test_activities.py | 2 + .../orchestrator/activities/test_mongo_db.py | 48 ++++++++++++++++++- .../utils/test_connectors_config.py | 4 ++ 8 files changed, 79 insertions(+), 6 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 5ab46c1..c785e7b 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. +- **`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/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/activities.py b/orchestrator/activities/activities.py index 04d1c36..f9d1228 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -84,6 +84,7 @@ class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postg connection_string=mongodb_config['connection_string'], database_name=mongodb_config['database_name'], ttl_index_seconds=mongodb_config['ttl_index_seconds'], + notification_queue_batch_size=mongodb_config['notification_queue_batch_size'], logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 51606fa..c3c8972 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -31,6 +31,8 @@ class MongoDB(SientiaMonitoring): ttl_index_seconds (int): TTL index duration in seconds logger (Logger): Application logger instance notification_handler (NotificationHandler): Notification management handler + notification_queue_batch_size (int): Hard cap on documents fetched per + `load_latest_data` run (SIENTIAPDE-2013). `0` means no limit. """ def __init__( @@ -41,6 +43,7 @@ class MongoDB(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, + notification_queue_batch_size: int = 0, ): SientiaMonitoring.__init__( self, @@ -61,6 +64,7 @@ class MongoDB(SientiaMonitoring): ) self.ttl_index_seconds = ttl_index_seconds + self.notification_queue_batch_size = notification_queue_batch_size # Initialize MongoDB client here (omitted for brevity) logger.info('MongoDB connection initialized') @@ -447,6 +451,16 @@ 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. + Args: input_data (dict[str, Any]): Activity input parameters. Required fields: @@ -488,7 +502,13 @@ class MongoDB(SientiaMonitoring): self.debug(f'Data filter: {data_filter}', metadata=metadata) - data = self.mongo_db_repository.find(collection_name, data_filter, metadata) + data = self.mongo_db_repository.find( + collection_name, + data_filter, + metadata, + limit=self.notification_queue_batch_size, + sort=[('timestamp', 1)], + ) self.debug(f'Collected: {data}', metadata=metadata) diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 2e1466c..14b2747 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -21,7 +21,8 @@ def build_mongodb_config(): Build MongoDB configuration from environment variables. Returns: - dict: MongoDB configuration with connection string, database name, and TTL index seconds. + dict: MongoDB configuration with connection string, database name, TTL index seconds, + and the notification queue's per-run batch size cap. """ username = getenv('MONGODB_USERNAME', 'root') password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c') @@ -32,6 +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')), } diff --git a/requirements.txt b/requirements.txt index 680f120..00de282 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -sientia_do>=1.12.2 +sientia_do>=1.13.0 prometheus-client diff --git a/tests/orchestrator/activities/test_activities.py b/tests/orchestrator/activities/test_activities.py index f9d979b..d006fb9 100644 --- a/tests/orchestrator/activities/test_activities.py +++ b/tests/orchestrator/activities/test_activities.py @@ -27,6 +27,7 @@ def test___init__( 'connection_string': 'mongodb://localhost:27017', 'database_name': 'test_db', 'ttl_index_seconds': 3600, + 'notification_queue_batch_size': 5000, } redis_config = {'host': 'localhost', 'port': 6379, 'username': 'admin', 'password': 'password'} @@ -89,6 +90,7 @@ def test___init__( connection_string='mongodb://localhost:27017', database_name='test_db', ttl_index_seconds=3600, + notification_queue_batch_size=5000, logger=logger, notification_handler=notification_handler, metrics_controller=mock_metrics_controller.return_value, diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 927f860..e1496b1 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -31,13 +31,14 @@ def test___init__(mongo_mock): notification_handler = MagicMock() metrics_controller = AsyncMock() - MongoDB( + mongo = MongoDB( connection_string='mongodb://localhost:27017', database_name='test_db', ttl_index_seconds=3600, logger=logger, notification_handler=notification_handler, metrics_controller=metrics_controller, + notification_queue_batch_size=5000, ) mongo_mock.assert_called_once_with( @@ -47,6 +48,22 @@ def test___init__(mongo_mock): notification_handler=notification_handler, metrics_controller=metrics_controller, ) + assert mongo.notification_queue_batch_size == 5000 + + +@patch('orchestrator.activities.mongo_db.MongoDBRepository') +def test___init___notification_queue_batch_size_defaults_to_zero(mongo_mock): + """0 means no limit, matching the underlying find()'s own convention.""" + mongo = MongoDB( + connection_string='mongodb://localhost:27017', + database_name='test_db', + ttl_index_seconds=3600, + logger=MagicMock(), + notification_handler=MagicMock(), + metrics_controller=AsyncMock(), + ) + + assert mongo.notification_queue_batch_size == 0 @patch('orchestrator.activities.mongo_db.SientiaMonitoring') @@ -480,6 +497,8 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db): 'test_collection', {'level': 'ERROR'}, {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + limit=0, + sort=[('timestamp', 1)], ) assert result == [ @@ -487,6 +506,29 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db): ] +def test_load_latest_data_passes_configured_batch_size_as_limit(mongo_db): + """SIENTIAPDE-2013: the configured notification_queue_batch_size caps the query.""" + mongo_db.notification_queue_batch_size = 100 + mongo_db.mongo_db_repository.find = MagicMock(return_value=[]) + + mongo_db.load_latest_data( + { + 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + 'collection_name': 'test_collection', + 'last_data_timestamp': None, + 'base_data_filter': {'level': 'ERROR'}, + } + ) + + mongo_db.mongo_db_repository.find.assert_called_once_with( + 'test_collection', + {'level': 'ERROR'}, + {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + limit=100, + sort=[('timestamp', 1)], + ) + + def test_load_latest_data_not_none_last_data_timestamp(mongo_db): """Test load_latest_data compares native datetime against native datetime in $gt""" @@ -516,6 +558,8 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db): 'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)}, }, {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, + limit=0, + sort=[('timestamp', 1)], ) assert result == [ @@ -529,7 +573,7 @@ def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db): boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC) - def fake_find(collection_name, data_filter, metadata): + def fake_find(collection_name, data_filter, metadata, limit=0, sort=None): gt = data_filter.get('timestamp', {}).get('$gt') all_docs = [ { diff --git a/tests/orchestrator/utils/test_connectors_config.py b/tests/orchestrator/utils/test_connectors_config.py index 33d982b..8409d0b 100644 --- a/tests/orchestrator/utils/test_connectors_config.py +++ b/tests/orchestrator/utils/test_connectors_config.py @@ -64,10 +64,12 @@ def test_build_mongo_db_config_with_env_vars(): environ['MONGODB_URL'] = 'localhost:27018' environ['MONGODB_DATABASE_NAME'] = 'test_db' environ['MONGODB_TTL_INDEX_HOURS'] = '2' + environ['NOTIFICATION_QUEUE_BATCH_SIZE'] = '2000' assert build_mongodb_config() == { 'connection_string': 'mongodb://sientia1:sientia1@localhost:27018', 'database_name': 'test_db', 'ttl_index_seconds': 7200, + 'notification_queue_batch_size': 2000, } @@ -77,10 +79,12 @@ def test_build_mongo_db_config_with_defaults(): environ.pop('MONGODB_DATABASE_NAME', None) environ.pop('MONGODB_URL', None) environ.pop('MONGODB_TTL_INDEX_HOURS', None) + environ.pop('NOTIFICATION_QUEUE_BATCH_SIZE', None) assert build_mongodb_config() == { 'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018', 'database_name': 'sientia', 'ttl_index_seconds': 3600, + 'notification_queue_batch_size': 5000, }