From 6479737dfb2130da788cfe1156a2442829110e06 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Tue, 15 Sep 2026 11:13:17 -0300 Subject: [PATCH] fix: harden ignore_models_id against explicit null, document levels-union design Code review findings on the SIENTIAPDE-2011 PR: - [Major] _compute_query_time_filter used group.get('ignore_models_id', []) for the intersection, whose default only applies when the key is missing. receiver_groups has no schema validation, so a document with ignore_models_id explicitly set to null (key present, value None) made .get() return None, and set(None) raised TypeError, crashing the whole activity for every mail type. Fixed with `group.get(...) or []`. Added a regression test. - [Minor] CLAUDE.md:48 described Alerts' base data filter as always `level: ERROR`, which this ticket's own query-time union can now widen. Updated the line to reflect that. - [Blocker, not applied] A finding argued the levels union should AND with default_filter's own level restriction instead of replacing it, to avoid widening Alerts (ERROR-only real-time channel) when a receiver_group sets broader levels. Both SIENTIAPDE-2011's ticket text and the parent design doc explicitly call for the union to replace the hard-coded ERROR-only default ("whatever levels the active groups actually care about"), so applying that fix would revert the subtask's actual purpose. The interim gap the finding correctly identifies (no per-group level filter exists yet until SIENTIAPDE-2012 lands) is now documented in _compute_query_time_filter's docstring instead of "fixed" by narrowing query-time behavior the ticket asked for. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 2 +- .../subworkflows/load_notification_package.py | 16 ++++++++++++++- .../test_load_notification_package.py | 20 +++++++++++++++++++ 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 66e7092..5ab46c1 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` vs all levels). +- **`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/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/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 0f28ec2..614899c 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -122,6 +122,17 @@ class LoadNotificationPackage: restriction (missing/empty `levels`), that group wants every level, so the query can't restrict levels at all -> falls back to whatever level restriction `default_filter` itself carries (e.g. ERROR-only for Alerts). + + When every active group DOES restrict `levels`, the union REPLACES + `default_filter`'s own level restriction rather than ANDing with it — + this is deliberate, not an oversight: SIENTIAPDE-2011's design doc and + ticket text both call for Alerts' hard-coded `level: ERROR` to become + "whatever levels the active groups actually care about", falling back + to ERROR-only only when no group narrows further. Until SIENTIAPDE-2012 + (per-group `levels` filtering) lands, this means a single active group + configuring `levels` to include WARNING/INFO widens the Alerts fetch for + every active group, not just that one — an accepted interim gap, not a + bug in this function. - ignore_models_id: intersection of every group's `ignore_models_id`. A model is only safe to drop from the fetch if EVERY active group ignores it. A group with no `ignore_models_id` is treated as "ignores nothing" (empty @@ -141,8 +152,11 @@ class LoadNotificationPackage: else: query_filter = dict(default_filter) + # `or []`, not a `.get(..., [])` default: a document with `ignore_models_id` + # explicitly set to `null` (no schema validation on this collection) makes + # `.get()` return None, and `set(None)` would raise TypeError. sending_configs_ignore_models_id = [ - set(group.get('ignore_models_id', [])) for group in sending_configs + set(group.get('ignore_models_id') or []) for group in sending_configs ] models_intersection = set.intersection(*sending_configs_ignore_models_id) diff --git a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py index 9f7d01b..4f049be 100644 --- a/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py +++ b/tests/orchestrator/workflows/subworkflows/test_load_notification_package.py @@ -233,3 +233,23 @@ def test_compute_query_time_filter_union_levels_and_intersection_models( 'level': {'$in': ['ERROR', 'WARNING']}, 'model_id': {'$nin': ['model_a']}, } + + +def test_compute_query_time_filter_ignore_models_id_explicit_none_does_not_raise( + load_notification_package, +): + # receiver_groups has no schema validation, so a document can have + # ignore_models_id explicitly set to null (key present, value None) rather + # than omitted. `group.get('ignore_models_id', [])` would still return None + # in that case (the default only applies when the key is missing), and + # `set(None)` raises TypeError -- must not crash the whole activity. + sending_configs = [ + {'group_name': 'group_a', 'levels': ['ERROR'], 'ignore_models_id': None}, + {'group_name': 'group_b', 'levels': ['ERROR'], 'ignore_models_id': ['model_a']}, + ] + default_filter = {} + + result = load_notification_package._compute_query_time_filter(sending_configs, default_filter) + + assert result == {'level': {'$in': ['ERROR']}} + assert 'model_id' not in result