Repository: rucio/rucio
Issue: `transfer-queued` message does not include the correct services

Motivation
----------
With all messages handled by the new hermes2 logic, messages get attached to the services defined in the configuration. This is then always handled when calling `add_message`. However, the `queue_requests` workflow does not call `add_message` but instead bulk inserts the messages to the database, not respecting the services logic.

https://github.com/rucio/rucio/blob/a5687ef8469e62e6466c8c26deaf306b3c0d7c9a/lib/rucio/core/request.py#L323-L324

Modification
------------
Either `add_message` needs to be called here, or better create a bulk logic for `add_message` which can be called here to have the advantage of the bulk inserts.

Changed files:
lib/rucio/core/request.py, lib/rucio/tests/test_message.py, lib/rucio/core/message.py

diff --git a/lib/rucio/core/message.py b/lib/rucio/core/message.py
index bc12989c8..2a1249c9a 100644
--- a/lib/rucio/core/message.py
+++ b/lib/rucio/core/message.py
@@ -15,55 +15,91 @@
 
 import json
 
+from typing import TYPE_CHECKING
+
 from sqlalchemy import or_, delete, update
 from sqlalchemy.exc import IntegrityError
 
 from rucio.common.constants import HermesService, MAX_MESSAGE_LENGTH
 from rucio.common.config import config_get_list
 from rucio.common.exception import InvalidObject, RucioException
-from rucio.common.utils import APIEncoder
+from rucio.common.utils import APIEncoder, chunks
 from rucio.db.sqla import filter_thread_work
 from rucio.db.sqla.models import Message, MessageHistory
 from rucio.db.sqla.session import transactional_session
 
+if TYPE_CHECKING:
+    from typing import Any, Dict, List, Optional
+    from sqlalchemy.orm import Session
+
+    MessageType = Dict[str, Any]
+    MessagesListType = List[MessageType]
+
 
 @transactional_session
-def add_message(event_type, payload, session=None):
+def add_messages(messages: "MessagesListType", session: "Optional[Session]" = None) -> None:
     """
-    Add a message to be submitted asynchronously to a message broker.
+    Add a list of messages to be submitted asynchronously to a message broker.
 
+    For messages with payload bigger than MAX_MESSAGE_LENGTH, payload_nolimit is used instead of payload.
     In the case of nolimit, a placeholder string is written to the NOT NULL payload column.
 
-    :param event_type: The type of the event as a string, e.g., NEW_DID.
-    :param payload: The message payload. Will be persisted as JSON.
+    :param messages: A list of dictionaries {'event_type': str, 'payload': dict}
     :param session: The database session to use.
     """
-    try:
-        payload = json.dumps(payload, cls=APIEncoder)
-    except TypeError as err:  # noqa: F841
-        raise InvalidObject('Invalid JSON for payload: %(err)s' % locals())
-
+    services = []
     for service in config_get_list('hermes', 'services_list', raise_exception=False, default='activemq,email', session=session):
         try:
             HermesService(service.upper())
         except ValueError as err:
             raise RucioException(str(err))
-        if event_type == 'email' and service != 'email':
-            continue
-        if service == 'email' and event_type != 'email':
-            continue
+        services.append(service)
+
+    msgs = []
+    for message in messages:
+        event_type = message['event_type']
+        payload = message['payload']
+        for service in services:
+            msg = {'services': service, 'event_type': event_type}
+            try:
+                if event_type == 'email' and service != 'email':
+                    continue
+                if service == 'email' and event_type != 'email':
+                    continue
+                msg_payload = json.dumps(payload, cls=APIEncoder)
+                msg['payload'] = msg_payload
+                if len(msg_payload) > MAX_MESSAGE_LENGTH:
+                    msg['payload_nolimit'] = msg_payload
+                    msg['payload'] = 'nolimit'
+                msgs.append(msg)
+            except TypeError as err:  # noqa: F841
+                raise InvalidObject('Invalid JSON for payload: %(err)s' % locals())
+    for messages_chunk in chunks(msgs, 1000):
+        session.bulk_insert_mappings(Message, messages_chunk)
 
-        if len(payload) > MAX_MESSAGE_LENGTH:
-            new_message = Message(event_type=event_type, payload='nolimit', payload_nolimit=payload, services=service)
-        else:
-            new_message = Message(event_type=event_type, payload=payload, services=service)
 
-        new_message.save(session=session, flush=False)
+@transactional_session
+def add_message(event_type: str, payload: dict, session: "Optional[Session]" = None) -> None:
+    """
+    Add a message to be submitted asynchronously to a message broker.
+
+    In the case of nolimit, a placeholder string is written to the NOT NULL payload column.
+
+    :param event_type: The type of the event as a string, e.g., NEW_DID.
+    :param payload: The message payload. Will be persisted as JSON.
+    :param session: The database session to use.
+    """
+    add_messages([{'event_type': event_type, 'payload': payload}], session=session)
 
 
 @transactional_session
-def retrieve_messages(bulk=1000, thread=None, total_threads=None, event_type=None,
-                      lock=False, old_mode=True, session=None):
+def retrieve_messages(bulk: int = 1000,
+                      thread: "Optional[int]" = None,
+                      total_threads: "Optional[int]" = None,
+                      event_type: "Optional[str]" = None,
+                      lock: bool = False,
+                      old_mode: bool = True,
+                      session: "Optional[Session]" = None) -> "MessagesListType":
     """
     Retrieve up to $bulk messages.
 
@@ -139,7 +175,7 @@ def retrieve_messages(bulk=1000, thread=None, total_threads=None, event_type=Non
 
 
 @transactional_session
-def delete_messages(messages, session=None):
+def delete_messages(messages: "MessagesListType", session: "Optional[Session]" = None) -> None:
     """
     Delete all messages with the given IDs, and archive them to the history.
 
@@ -165,7 +201,7 @@ def delete_messages(messages, session=None):
 
 
 @transactional_session
-def truncate_messages(session=None):
+def truncate_messages(session: "Optional[Session]" = None) -> None:
     """
     Delete all stored messages. This is for internal purposes only.
 
@@ -179,7 +215,7 @@ def truncate_messages(session=None):
 
 
 @transactional_session
-def update_messages_services(messages, services, session=None):
+def update_messages_services(messages: "MessagesListType", services: str, session: "Optional[Session]" = None) -> None:
     """
     Update the services for all messages with the given IDs.
 
diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index 55ac09536..2a81e01e9 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -30,7 +30,7 @@ from rucio.common.config import config_get_bool, config_get
 from rucio.common.exception import RequestNotFound, RucioException, UnsupportedOperation
 from rucio.common.types import InternalAccount, InternalScope
 from rucio.common.utils import generate_uuid, chunks, get_parsed_throttler_mode
-from rucio.core.message import add_message
+from rucio.core.message import add_message, add_messages
 from rucio.core.monitor import record_counter, record_timer
 from rucio.core.rse import get_rse_name, get_rse_vo, get_rse_transfer_limits, get_rse_attribute, RseData
 from rucio.db.sqla import models, filter_thread_work
@@ -313,7 +313,7 @@ def queue_requests(requests, session=None, logger=logging.log):
                    'queued_at': str(datetime.datetime.utcnow())}
 
         messages.append({'event_type': transfer_status,
-                         'payload': json.dumps(payload)})
+                         'payload': payload})
 
     for requests_chunk in chunks(new_requests, 1000):
         session.bulk_insert_mappings(models.Request, requests_chunk)
@@ -321,8 +321,7 @@ def queue_requests(requests, session=None, logger=logging.log):
     for sources_chunk in chunks(sources, 1000):
         session.bulk_insert_mappings(models.Source, sources_chunk)
 
-    for messages_chunk in chunks(messages, 1000):
-        session.bulk_insert_mappings(models.Message, messages_chunk)
+    add_messages(messages, session=session)
 
     return new_requests
 
diff --git a/lib/rucio/tests/test_message.py b/lib/rucio/tests/test_message.py
index f0d8096af..2910face4 100644
--- a/lib/rucio/tests/test_message.py
+++ b/lib/rucio/tests/test_message.py
@@ -13,10 +13,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
+import random
+import string
+
 import pytest
 
-from rucio.common.exception import InvalidObject
-from rucio.core.message import add_message, retrieve_messages, delete_messages, truncate_messages
+from rucio.db.sqla.models import Message
+from rucio.db.sqla.session import get_session
+from rucio.common.constants import MAX_MESSAGE_LENGTH
+from rucio.common.exception import InvalidObject, RucioException
+from rucio.common.utils import generate_uuid
+from rucio.core.message import add_message, add_messages, retrieve_messages, delete_messages, truncate_messages
 
 
 @pytest.mark.noparallel(reason='fails when run in parallel')
@@ -48,35 +56,92 @@ def test_add_message(core_config_mock, caches_mock):
 @pytest.mark.parametrize("caches_mock", [{"caches_to_mock": [
     'rucio.core.config.REGION',
 ]}], indirect=True)
-def test_pop_messages(core_config_mock, caches_mock):
-    """ MESSAGE (CORE): Test retrieve and delete messages """
+def test_bulk_insert_and_pop_messages(core_config_mock, caches_mock):
+    """ MESSAGE (CORE): Test bulk insert, retrieve and delete messages """
 
     truncate_messages()
     list_messages = []
+    messages = []
     for cnt in range(10):
-        add_message(event_type='TEST', payload={'foo': True,
-                                                'monty': 'python',
-                                                'number': cnt})
-        list_messages.append((cnt, 'influx'))
-        list_messages.append((cnt, 'activemq'))
-        list_messages.append((cnt, 'elastic'))
-
-    messages = retrieve_messages(30)
+        messages.append(
+            {
+                "event_type": generate_uuid()[:10],
+                "payload": {"foo": True, "monty": "python", "number": cnt},
+            }
+        )
+    add_messages(messages)
+
+    list_messages = retrieve_messages(40)
+    assert len(list_messages) == 30
     to_delete = []
     for message in messages:
-        assert isinstance(message['payload'], dict)
-        assert message['payload']['foo'] is True
-        assert message['payload']['monty'] == 'python'
-        assert message['payload']['number'] in list(range(10))
-        to_delete.append({'id': message['id'],
-                          'created_at': message['created_at'],
-                          'updated_at': message['created_at'],
-                          'payload': str(message['payload']),
-                          'event_type': message['event_type']})
-        list_messages.remove((message['payload']['number'], message['services']))
-
-    assert list_messages == []
-
+        filtered_messages = [msg for msg in list_messages if msg['event_type'] == message['event_type']]
+        assert len(filtered_messages) == 3
+        services = ['influx', 'activemq', 'elastic']
+        for msg in filtered_messages:
+            assert isinstance(msg['payload'], dict)
+            assert msg['payload']['foo'] is True
+            assert msg['payload']['monty'] == 'python'
+            assert msg['payload']['number'] in list(range(10))
+            assert msg['services'] in services
+            services.remove(msg['services'])
+            to_delete.append(
+                {
+                    "id": msg["id"],
+                    "created_at": msg["created_at"],
+                    "updated_at": msg["created_at"],
+                    "payload": str(msg["payload"]),
+                    "event_type": msg["event_type"],
+                }
+            )
     delete_messages(to_delete)
 
     assert retrieve_messages() == []
+
+
+@pytest.mark.noparallel(reason='fails when run in parallel')
+@pytest.mark.parametrize("core_config_mock", [{"table_content": [
+    ('hermes', 'services_list', 'influx,activemq,elastic,email'),
+]}], indirect=True)
+@pytest.mark.parametrize("caches_mock", [{"caches_to_mock": [
+    'rucio.core.config.REGION',
+]}], indirect=True)
+def test_large_payload(core_config_mock, caches_mock):
+    """ MESSAGE (CORE): Test insert and retrieving of message with large payload """
+    truncate_messages()
+
+    long_payload = ''.join(random.choice(string.ascii_letters) for i in range(MAX_MESSAGE_LENGTH + 20))
+    dict_long_payload = {"mylong_message": long_payload}
+    event_type = generate_uuid()[:10]
+    add_message(event_type=event_type, payload=dict_long_payload)
+
+    session = get_session()
+    msg = session.query(Message.id,   # pylint: disable=no-member
+                        Message.created_at,
+                        Message.event_type,
+                        Message.payload,
+                        Message.payload_nolimit,
+                        Message.services)\\
+        .filter_by(event_type=event_type)\\
+        .first()
+    assert msg.payload == 'nolimit'
+    assert msg.payload_nolimit == json.dumps(dict_long_payload)
+    messages = retrieve_messages(40)
+    assert messages[0]['payload'] == dict_long_payload
+
+
+@pytest.mark.noparallel(reason='fails when run in parallel')
+@pytest.mark.parametrize("core_config_mock", [{"table_content": [
+    ('hermes', 'services_list', 'nonexistingservice'),
+]}], indirect=True)
+@pytest.mark.parametrize("caches_mock", [{"caches_to_mock": [
+    'rucio.core.config.REGION',
+]}], indirect=True)
+def test_non_existing_service(core_config_mock, caches_mock):
+    """ MESSAGE (CORE): Test insert message to a non-existing service """
+    truncate_messages()
+
+    with pytest.raises(RucioException):
+        add_message(event_type='NEW_DID', payload={'name': 'name',
+                                                   'name_Y': 'scope_X',
+                                                   'type': 'file'})