improve async check

tests/integration/server_rest_api/datex2/datex2_import_api_test.py

@@ -337,6 +337,48 @@ def test_push_realtime_v35_large_delta_push_is_async(
         evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first()
         assert evse_charging.status == EvseStatus.UNKNOWN
 
+    @staticmethod
+    def test_push_realtime_v35_oversized_payload_skips_validation_and_queues(
+        db: SQLAlchemy,
+        test_client: OpenApiFlaskClient,
+        requests_mock: Mocker,
+        isolated_datex2_dir: Path,
+        stubbed_celery_delay,
+    ) -> None:
+        """
+        Payloads larger than DATEX2_SYNC_MAX_CONTENT_LENGTH must be handed off to celery without
+        being parsed or validated on the request thread - so even a structurally invalid payload is
+        persisted + queued (HTTP 200) instead of being rejected with HTTP 400.
+        """
+        _import_static_data(requests_mock)
+
+        # A payload that would normally be rejected as invalid, but is pushed over the size limit.
+        invalid_payload = {'not': 'a valid messageContainer'}
+
+        config = dependencies.get_config_helper().get_config()
+        sentinel = object()
+        original = config.get('DATEX2_SYNC_MAX_CONTENT_LENGTH', sentinel)
+        config['DATEX2_SYNC_MAX_CONTENT_LENGTH'] = 5
+        try:
+            response = test_client.post(
+                path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}',
+                json=invalid_payload,
+            )
+        finally:
+            if original is sentinel:
+                config.pop('DATEX2_SYNC_MAX_CONTENT_LENGTH', None)
+            else:
+                config['DATEX2_SYNC_MAX_CONTENT_LENGTH'] = original
+
+        assert response.status_code == HTTPStatus.OK
+
+        # Handed off to celery unvalidated: payload persisted to disk and task queued.
+        written = list(isolated_datex2_dir.iterdir())
+        assert len(written) == 1
+        with written[0].open('rb') as f:
+            assert json.load(f) == invalid_payload
+        stubbed_celery_delay.assert_called_once()
+
     @staticmethod
     def test_push_realtime_v35_invalid_payload_returns_400(
         db: SQLAlchemy,

webapp/common/constants.py

@@ -42,6 +42,10 @@ class BaseConfig:
     # processed asynchronously via celery instead of inline on the request thread (snapshots are always async).
     DATEX2_ASYNC_STATION_STATUS_THRESHOLD = 25
 
+    # DATEX II realtime pushes with a request body larger than this many bytes are handed off to celery
+    # without being parsed or validated on the request thread, so large payloads never block it.
+    DATEX2_SYNC_MAX_CONTENT_LENGTH = 250 * 1024
+
     PUBLIC_IMAGE_PATH = '/data/images'
 
     SQLALCHEMY_TRACK_MODIFICATIONS = False

webapp/server_rest_api/datex2/datex2_rest_api.py

@@ -104,12 +104,14 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         summary='Push DATEX II v3.5 realtime EVSE statuses',
         description=(
             'Accepts a Mobilithek-style DATEX II v3.5 JSON payload (`messageContainer.payload[0]` '
-            '`aegiEnergyInfrastructureStatusPublication`). The payload is validated synchronously; '
-            'small `deltaPush` updates are applied directly, while snapshots (any other exchange '
-            'protocol) and `deltaPush` payloads with more than `DATEX2_ASYNC_STATION_STATUS_THRESHOLD` '
-            '`energyInfrastructureStationStatus` entries are persisted to `DATEX2_IMPORT_DIR` and '
-            'processed asynchronously by a Celery worker so very large payloads do not block the '
-            "request thread. The `key` query parameter must match the source's configured `api_key`."
+            '`aegiEnergyInfrastructureStatusPublication`). Payloads larger than '
+            '`DATEX2_SYNC_MAX_CONTENT_LENGTH` bytes are persisted to `DATEX2_IMPORT_DIR` and processed '
+            'asynchronously by a Celery worker without being parsed or validated on the request thread. '
+            'Smaller payloads are validated synchronously; small `deltaPush` updates are applied '
+            'directly, while snapshots (any other exchange protocol) and `deltaPush` payloads with more '
+            'than `DATEX2_ASYNC_STATION_STATUS_THRESHOLD` `energyInfrastructureStationStatus` entries are '
+            'persisted and processed asynchronously as well. The `key` query parameter must match the '
+            "source's configured `api_key`."
         ),
         path=[Parameter('source_uid', schema=StringField())],
         query=[
@@ -138,6 +140,15 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         if not data:
             raise InputValidationException(message='no realtime payload')
 
+        # Parsing and validating a large payload synchronously is expensive and would block the
+        # request thread, so anything above DATEX2_SYNC_MAX_CONTENT_LENGTH bytes is handed off to a
+        # celery worker unparsed and unvalidated - the worker parses, validates and applies it.
+        sync_max_content_length = self.config_helper.get('DATEX2_SYNC_MAX_CONTENT_LENGTH', 250 * 1024)
+        if len(data) > sync_max_content_length:
+            self._queue_for_async_processing(source_uid, data)
+            # Mobilithek expects HTTP 200 instead of HTTP 204
+            return empty_json_response(), HTTPStatus.OK
+
         try:
             parsed_data = json.loads(data)
         except json.JSONDecodeError as e:
@@ -158,6 +169,16 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
             # Mobilithek expects HTTP 200 instead of HTTP 204
             return empty_json_response(), HTTPStatus.OK
 
+        self._queue_for_async_processing(source_uid, data)
+
+        # Mobilithek expects HTTP 200 instead of HTTP 204
+        return empty_json_response(), HTTPStatus.OK
+
+    def _queue_for_async_processing(self, source_uid: str, data: bytes) -> None:
+        """
+        Persist a raw realtime payload to ``DATEX2_IMPORT_DIR`` and hand it off to a celery worker
+        for asynchronous parsing, validation and application via ``datex2_v3_5_realtime_import_by_file``.
+        """
         base_path = Path(self.config_helper.get('DATEX2_IMPORT_DIR'))
         if not base_path.is_dir():
             base_path.mkdir(parents=True, exist_ok=True)
@@ -168,9 +189,6 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
 
         self.celery_helper.delay(datex2_v3_5_realtime_import_by_file, source_uid, str(import_path))
 
-        # Mobilithek expects HTTP 200 instead of HTTP 204
-        return empty_json_response(), HTTPStatus.OK
-
     def _dump_request(self, source_uid: str, request_body: bytes) -> None:
         source_config = self.config_helper.get('SOURCES', {}).get(source_uid) or {}
         if source_config.get('debug', False) is False: