SIENTIAPDE-2007 SIENTIAPDE-2007: Compare native datetime in load_latest_data $gt filter and bump sientia_do pin to 1.12.2

Notification.timestamp is now stored as a native BSON Date instead of a
string, so load_latest_data must parse the ISO string coming from Redis
back to datetime before building the $gt filter, or the comparison
silently breaks (BSON String vs Date always evaluates False). Also fixes
put_last_data_timestamp, which would otherwise raise TypeError trying to
json.dumps a pandas Timestamp when persisting the new watermark to Redis.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

orchestrator/activities/mongo_db.py

@@ -2,7 +2,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     import traceback
-    from datetime import UTC
+    from datetime import UTC, datetime
     from logging import Logger
     from typing import Any
 
@@ -452,7 +452,8 @@ class MongoDB(SientiaMonitoring):
                 Required fields:
                 - metadata (dict[str, Any]): Workflow execution metadata
                 - collection_name (str): Name of the MongoDB collection
-                - last_data_timestamp (str | None): Last processed timestamp for filtering
+                - last_data_timestamp (str | None): Last processed timestamp
+                  (ISO-format string) for filtering
                 - base_data_filter (dict[str, Any]): Base query filter conditions
 
         Returns:
@@ -472,19 +473,14 @@ class MongoDB(SientiaMonitoring):
             if last_data_timestamp is None:
                 data_filter = base_data_filter
             else:
-                # ``notification_queue.timestamp`` is stored as a string in
+                # ``notification_queue.timestamp`` is stored as a native BSON
-                # ``DATETIME_FORMAT_WITH_TZ`` (``Notification`` writes it as
+                # ``Date`` (``Notification.timestamp`` is a ``datetime``).
-                # ``now().strftime(DATETIME_FORMAT_WITH_TZ)``). Coercing
+                # ``last_data_timestamp`` arrives here as an ISO-format string
-                # ``last_data_timestamp`` to ``datetime`` here would force a
+                # (round-tripped through Redis), so it must be parsed back to
-                # BSON ``String`` vs ``Date`` comparison, which always yields
+                # ``datetime`` for the ``$gt`` comparison to be type-correct.
-                # ``False`` (``String < Date`` in BSON sort order) and breaks
-                # incremental loading entirely. Comparing strings preserves the
-                # intended chronological filter because the format is
-                # lexicographically ordered when the timezone is fixed
-                # (``Notification.timestamp`` always uses UTC).
                 data_filter = {
                     **base_data_filter,
-                    'timestamp': {'$gt': last_data_timestamp},
+                    'timestamp': {'$gt': datetime.fromisoformat(last_data_timestamp)},
                 }
 
             self.debug(f'Data filter: {data_filter}', metadata=metadata)

orchestrator/activities/slot_manager.py

@@ -335,6 +335,11 @@ class SlotManager(SientiaMonitoring):
 
         self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
 
+        # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from
+        # Mongo), which the Redis repository's plain ``json.dumps`` cannot
+        # serialize. Store it as an ISO-format string instead.
+        last_data_timestamp = last_data_timestamp.isoformat()
+
         try:
             self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
         except Exception as e:

requirements-local.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
 prometheus-client

requirements.txt

@@ -4,5 +4,5 @@ sqlalchemy
 redis
 pymongo
 jinja2
-sientia_do>=1.12.1
+sientia_do>=1.12.2
 prometheus-client

tests/orchestrator/activities/test_mongo_db.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import UTC, datetime
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
 from pytest import fixture
@@ -462,7 +462,7 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
             {
                 'name': 'test1',
                 'value': 1,
-                'timestamp': '2023-01-01 12:00:00+0000',
+                'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC),
             }
         ]
     )
@@ -482,18 +482,20 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
     )
 
-    assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
+    assert result == [
+        {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)}
+    ]
 
 
 def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
-    """Test load_latest_data"""
+    """Test load_latest_data compares native datetime against native datetime in $gt"""
 
     mongo_db.mongo_db_repository.find = MagicMock(
         return_value=[
             {
                 'name': 'test1',
                 'value': 1,
-                'timestamp': '2023-01-01 12:00:00+0000',
+                'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC),
             }
         ]
     )
@@ -502,7 +504,7 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
         {
             'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
             'collection_name': 'test_collection',
-            'last_data_timestamp': '2023-01-01 12:00:00+0000',
+            'last_data_timestamp': '2023-01-01T12:00:00+00:00',
             'base_data_filter': {'level': 'ERROR'},
         }
     )
@@ -511,12 +513,56 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
         'test_collection',
         {
             'level': 'ERROR',
-            'timestamp': {'$gt': '2023-01-01 12:00:00+0000'},
+            'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)},
         },
         {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
     )
 
-    assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
+    assert result == [
+        {'name': 'test1', 'value': 1, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)}
+    ]
+
+
+def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db):
+    """Regression: the $gt filter built from the previous run's timestamp must not
+    skip the document that landed exactly on the boundary, nor re-return it."""
+
+    boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
+
+    def fake_find(collection_name, data_filter, metadata):
+        gt = data_filter.get('timestamp', {}).get('$gt')
+        all_docs = [
+            {'name': 'before', 'value': 1, 'timestamp': datetime(2023, 1, 1, 11, 59, 59, tzinfo=UTC)},
+            {'name': 'boundary', 'value': 2, 'timestamp': boundary_timestamp},
+            {'name': 'after', 'value': 3, 'timestamp': datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC)},
+        ]
+        if gt is None:
+            return all_docs
+        return [doc for doc in all_docs if doc['timestamp'] > gt]
+
+    mongo_db.mongo_db_repository.find = MagicMock(side_effect=fake_find)
+
+    first_run = 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'},
+        }
+    )
+
+    assert [doc['name'] for doc in first_run] == ['before', 'boundary', 'after']
+
+    second_run = mongo_db.load_latest_data(
+        {
+            'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
+            'collection_name': 'test_collection',
+            'last_data_timestamp': boundary_timestamp.isoformat(),
+            'base_data_filter': {'level': 'ERROR'},
+        }
+    )
+
+    assert [doc['name'] for doc in second_run] == ['after']
 
 
 def test_load_latest_data_error(mongo_db):
@@ -528,7 +574,7 @@ def test_load_latest_data_error(mongo_db):
             {
                 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
                 'collection_name': 'test_collection',
-                'last_data_timestamp': '2023-01-01 12:00:00+0000',
+                'last_data_timestamp': '2023-01-01T12:00:00+00:00',
                 'base_data_filter': {'level': 'ERROR'},
             }
         )

tests/orchestrator/activities/test_slot_manager.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pandas import DataFrame
@@ -279,7 +279,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
         {
             'name': ['sensor1', 'sensor2'],
             'value': [25.5, 30.0],
-            'timestamp': ['2023-01-01 12:00:00', '2023-01-01 12:00:01'],
+            'timestamp': [
+                datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC),
+                datetime(2023, 1, 1, 12, 0, 1, tzinfo=UTC),
+            ],
         }
     )
     test_data = {
@@ -294,10 +297,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
 
     result = slot_manager.put_last_data_timestamp(test_data)
 
-    assert result == '2023-01-01 12:00:01'
+    assert result == '2023-01-01T12:00:01+00:00'
 
     slot_manager.redis_repository.set.assert_called_once_with(
-        'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01', ttl=18000
+        'notification_last_timestamp:test_mail_type', '2023-01-01T12:00:01+00:00', ttl=18000
     )
 
 
@@ -311,7 +314,7 @@ def test_put_last_data_timestamp_error(slot_manager):
             {
                 'name': ['sensor1', 'sensor2'],
                 'value': [25.5, 30.0],
-                'timestamp': ['2023-01-01 12:00:00'] * 2,
+                'timestamp': [datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)] * 2,
             }
         ).to_dict('records'),
         'mail_type': 'test_mail_type',