FIX improve notification queue full log error including subscription id

CHANGES_NEXT_RELEASE

@@ -2,6 +2,7 @@
 - Fix: notification was not triggered in subs using entityUpdate alterationType when some attribute not included in condition.attrs has an actual value change (#4647)
 - Fix: previousValue metadata was not taken into account in attributes not in the update that triggers the notification (#4643)
 - Fix: improve attribute and metadata invalid format dates for DateTime types in logs (#4616)
+- Fix: improve notification queue full log error including subscription id
 - Fix: attrsFormat keyValues and values were not working in custom notifications using ngsi payloads (#4644)
 - Fix: JEXL expressions were not working in attrsFormat simplifiedKeyValues, keyValues or values (#4645)
 - Hardening: upgrade microhttpd dependency from 0.9.76 to 1.0.1

src/lib/common/SyncQOverflow.h

@@ -27,6 +27,7 @@
 */
 
 #include <queue>
+#include <map>
 
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/condition_variable.hpp>
@@ -49,6 +50,7 @@ class SyncQOverflow
     bool try_push(Data element, bool unstoppable = false);
     Data pop();
     size_t size() const;
+    std::string countSubIds() const;
 };
 
 /* ****************************************************************************
@@ -70,6 +72,38 @@ bool SyncQOverflow<Data>::try_push(Data element, bool unstoppable)
   return false;
 }
 
+/* ****************************************************************************
+*
+* SyncQOverflow<Data>::countSubIds -
+*/
+template <typename Data>
+std::string SyncQOverflow<Data>::countSubIds() const
+{
+  boost::mutex::scoped_lock lock(mtx);
+  std::map<std::string, size_t> occurrences;
+
+  // FIXME PR: unsure about this solution. Copying the full queue each time seem to be overkill....
+  std::queue<Data> tempQueue = queue;
+  while (!tempQueue.empty())
+  {
+    Data element = tempQueue.front();
+    ++occurrences[element->subscriptionId];
+    tempQueue.pop();
+  }
+
+  std::string result;
+  for (const auto& pair : occurrences)
+  {
+    if (!result.empty())
+    {
+      result += ", ";
+    }
+    result += pair.first + " (" + std::to_string(pair.second) + ")";
+  }
+
+  return result;
+}
+
 /* ****************************************************************************
 *
 * SyncQOverflow<Data>::pop -

src/lib/ngsiNotify/QueueNotifier.cpp

@@ -180,7 +180,7 @@ void QueueNotifier::sendNotifyContextRequest
   if (!enqueued)
   {
     QueueStatistics::incReject(1);
-    LM_E(("Runtime Error (%s notification queue is full)", queueName.c_str()));
+    LM_E(("Runtime Error (%s notification queue is full for sub %s, queue occupancy is: %s)", queueName.c_str(), nsf.subId.c_str(), sq->countSubIds().c_str()));
     delete paramsP;
 
     return;

src/lib/ngsiNotify/ServiceQueue.cpp

@@ -71,6 +71,15 @@ bool ServiceQueue::try_push(SenderThreadParams* item)
 }
 
 
+/* ****************************************************************************
+*
+* ServiceQueue::countSubIds -
+*/
+std::string ServiceQueue::countSubIds() const
+{
+  return queue.countSubIds();
+}
+
 
 /* ****************************************************************************
 *

src/lib/ngsiNotify/ServiceQueue.h

@@ -43,6 +43,8 @@ class ServiceQueue
   int stop(void);
   bool try_push(SenderThreadParams* item);
   size_t size() const;
+  std::string countSubIds() const;
+
 
 private:
   SyncQOverflow<SenderThreadParams*>  queue;

test/functionalTest/cases/4649_include_queue_snapshot_when_is_full/include_queue_snapshot_when_is_full.test

@@ -0,0 +1,300 @@
+# Copyright 2025 Telefonica Investigacion y Desarrollo, S.A.U
+#
+# This file is part of Orion Context Broker.
+#
+# Orion Context Broker is free software: you can redistribute it and/or
+# modify it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# Orion Context Broker is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
+# General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with Orion Context Broker. If not, see http://www.gnu.org/licenses/.
+#
+# For those usages not covered by this license please contact with
+# iot_support at tid dot es
+
+# VALGRIND_READY - to mark the test ready for valgrindTestSuite.sh
+
+--NAME--
+Log queue snapshot when quee is full
+
+--SHELL-INIT--
+dbInit CB
+brokerStart CB 0 IPv4 -notificationMode threadpool:6:6 -httpTimeout 20000
+accumulatorStart --pretty-print
+
+--SHELL--
+
+#
+# 01. Create sub1 for E1 using endpoint that nevers responses
+# 02. Create sub2 for E2 using endpoint that nevers responses
+# 03. Create sub3 for E2 using endpoint that nevers responses
+# 04. Trigger sub1 1 times
+# 05. Trigger sub2 2 times
+# 06. Trigger sub3 3 times
+# 07. Trigger sub1 one more time: queue is full
+# 08. Get log and see queue full for sub1, sub1(1), sub2(2), sub3(3)
+#
+
+echo "01. Create sub1 for E1 using endpoint that nevers responses"
+echo "==========================================================="
+payload='{
+  "subject": {
+    "entities": [
+      {
+        "id": "E1",
+        "type": "T"
+      }
+    ]
+  },
+  "notification": {
+    "http": {
+      "url": "http://127.0.0.1:'${LISTENER_PORT}'/waitForever"
+    }
+  }
+}'
+orionCurl --url /v2/subscriptions --payload "$payload"
+echo
+echo
+
+
+SUB1_ID=$(echo "$_response" | grep subscriptionId | awk -F '>' '{print $2}' | awk -F '<' '{print $1}' | grep -v '^$' )
+
+
+echo "02. Create sub2 for E2 using endpoint that nevers responses"
+echo "==========================================================="
+payload='{
+  "subject": {
+    "entities": [
+      {
+        "id": "E2",
+        "type": "T"
+      }
+    ]
+  },
+  "notification": {
+    "http": {
+      "url": "http://127.0.0.1:'${LISTENER_PORT}'/waitForever"
+    }
+  }
+}'
+orionCurl --url /v2/subscriptions --payload "$payload"
+echo
+echo
+
+
+SUB2_ID=$(echo "$_response" | grep subscriptionId | awk -F '>' '{print $2}' | awk -F '<' '{print $1}' | grep -v '^$' )
+
+
+echo "03. Create sub3 for E3 using endpoint that nevers responses"
+echo "==========================================================="
+payload='{
+  "subject": {
+    "entities": [
+      {
+        "id": "E3",
+        "type": "T"
+      }
+    ]
+  },
+  "notification": {
+    "http": {
+      "url": "http://127.0.0.1:'${LISTENER_PORT}'/waitForever"
+    }
+  }
+}'
+orionCurl --url /v2/subscriptions --payload "$payload"
+echo
+echo
+
+
+SUB3_ID=$(echo "$_response" | grep subscriptionId | awk -F '>' '{print $2}' | awk -F '<' '{print $1}' | grep -v '^$' )
+
+
+echo "04. Trigger sub1 1 times"
+echo "========================"
+payload='
+{
+  "id": "E1",
+  "type": "T",
+  "A": {
+    "type": "Number",
+    "value": 20
+  }
+}'
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+
+
+echo "05. Trigger sub2 2 times"
+echo "========================"
+payload='
+{
+  "id": "E2",
+  "type": "T",
+  "A": {
+    "type": "Number",
+    "value": 20
+  }
+}'
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+
+
+echo "06. Trigger sub3 3 times"
+echo "========================"
+payload='
+{
+  "id": "E3",
+  "type": "T",
+  "A": {
+    "type": "Number",
+    "value": 20
+  }
+}'
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+
+
+echo "07. Trigger sub1 one more time: queue is full"
+echo "============================================="
+payload='
+{
+  "id": "E1",
+  "type": "T",
+  "A": {
+    "type": "Number",
+    "value": 20
+  }
+}'
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+orionCurl --url /v2/entities?options=forcedUpdate,upsert --payload "$payload"
+echo
+echo
+
+
+echo "08. Get log and see queue full for sub1, sub1(1), sub2(2), sub3(3)"
+echo "=================================================================="
+cat /tmp/contextBroker.log | grep 'ERROR' | awk -F 'msg=' '{print $2}'
+echo
+echo
+
+
+--REGEXPECT--
+01. Create sub1 for E1 using endpoint that nevers responses
+===========================================================
+HTTP/1.1 201 Created
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/subscriptions/REGEX([0-9a-f]{24})
+Content-Length: 0
+
+
+
+02. Create sub2 for E2 using endpoint that nevers responses
+===========================================================
+HTTP/1.1 201 Created
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/subscriptions/REGEX([0-9a-f]{24})
+Content-Length: 0
+
+
+
+03. Create sub3 for E3 using endpoint that nevers responses
+===========================================================
+HTTP/1.1 201 Created
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/subscriptions/REGEX([0-9a-f]{24})
+Content-Length: 0
+
+
+
+04. Trigger sub1 1 times
+========================
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E1?type=T
+
+
+
+05. Trigger sub2 2 times
+========================
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E2?type=T
+
+
+
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E2?type=T
+
+
+
+06. Trigger sub3 3 times
+========================
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E3?type=T
+
+
+
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E3?type=T
+
+
+
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E3?type=T
+
+
+
+07. Trigger sub1 one more time: queue is full
+=============================================
+HTTP/1.1 204 No Content
+Date: REGEX(.*)
+Fiware-Correlator: REGEX([0-9a-f\-]{36})
+Location: /v2/entities/E1?type=T
+
+
+
+08. Get log and see queue full for sub1, sub1(1), sub2(2), sub3(3)
+==================================================================
+X
+
+
+--TEARDOWN--
+brokerStop CB
+dbDrop CB
+accumulatorStop