Fix various issues with RMQ publish confirmations

salcock · salcock · commit b916b40ec83f · 2025-10-03T19:10:03.000+13:00
* resolve bug where an ACK was being consumed by our "blocking"
   check
 * fix bug where integrity hash PDUs were not being included in the
   set of unconfirmed publishes
 * reduce ACK fetching timeout to a single second
 * fix bug where saved message counts were not properly reset
   if all of the messages of that type were ACKed but messages
   of another type remained unACKed.
 * do not destroy the RMQ socket if the ACK check times out, as we
   can just try again later
 * fix segfault due to continuing to process data or timer events
   after another event had caused us to tear down the RMQ
   connection
diff --git a/src/mediator/coll_recv_thread.c b/src/mediator/coll_recv_thread.c
@@ -36,6 +36,10 @@
 #include "med_epoll.h"
 #include "etsili_core.h"
 
+#define COLL_OUTSTANDING_PUB_CONFIRMS(col) \
+    (col->saved_iri_msg_cnt > 0 || col->saved_cc_msg_cnt > 0 || \
+            col->saved_raw_msg_cnt > 0)
+
 /** This file implements a "collector receive" thread for the OpenLI mediator.
  *  Each OpenLI collector that reports to a mediator will be handled using
  *  a separate instance of one of these threads.
@@ -1116,6 +1120,7 @@ static int receive_collector(coll_recv_t *col, med_epoll_ev_t *mev) {
 
 processacks:
     if (col->amqp_producer_state &&
+            COLL_OUTSTANDING_PUB_CONFIRMS(col) && 
             consume_mediator_RMQ_producer_acks(col) == 0) {
         /* RMQ failed to acknowledge everything we published, have to
          * reconnect and re-publish
@@ -1620,6 +1625,13 @@ static void *start_collector_thread(void *params) {
                     move_thread_into_error_state(col, 1);
                     break;
                 }
+                if (col->amqp_producer_state == NULL) {
+                    /* We've dropped our internal RMQ session -- no point
+                     * in continuing processing data / responding to timer
+                     * events until we've had a change to reconnect
+                     */
+                     break;
+                }
             }
         }
         /* If we get here, the message timer expired -- loop around and
diff --git a/src/mediator/coll_recv_thread.h b/src/mediator/coll_recv_thread.h
@@ -51,7 +51,7 @@
  *
  */
 
-#define MAX_SAVED_RECEIVED_DATA 32
+#define MAX_SAVED_RECEIVED_DATA 256
 
 /** Types of messages that can be sent between the main mediator thread and a
  *  collector receive thread.
diff --git a/src/mediator/mediator_integrity_check.c b/src/mediator/mediator_integrity_check.c
@@ -453,6 +453,12 @@ uint8_t send_integrity_check_hash_pdu(coll_recv_t *col,
 
     encres = generate_integrity_check_hash_pdu(ics, medid, operatorid,
             col->etsiencoder, col->etsidecoder);
+    if (collrecv_save_message(col, (unsigned char *)ics->liid, encres->encoded,
+            encres->len, ics->msgtype) < 0) {
+        wandder_release_encoded_result(col->etsiencoder, encres);
+        return -1;
+    }
+
     if (ics->msgtype == OPENLI_PROTO_ETSI_CC) {
         r = publish_cc_on_mediator_liid_RMQ_queue(col->amqp_producer_state,
                 encres->encoded, encres->len, found->liid,
@@ -522,6 +528,9 @@ int send_integrity_check_sign_pdu(coll_recv_t *col,
 
     if (collrecv_save_message(col, (unsigned char *)ics->liid, encres->encoded,
             encres->len, ics->msgtype) < 0) {
+        if (operatorid) {
+            free(operatorid);
+        }
         wandder_release_encoded_result(col->etsiencoder, encres);
         return -1;
     }
diff --git a/src/mediator/mediator_rmq.c b/src/mediator/mediator_rmq.c
@@ -146,70 +146,6 @@ static int register_RMQ_consumer(amqp_connection_state_t state,
     return 0;
 }
 
-static int update_mediator_rmq_connection_block_status(
-        amqp_connection_state_t state, uint8_t *is_blocked) {
-
-    /* copy of code from export_buffer.c, but with different log
-     * messages when things go awry
-     */
-    amqp_frame_t frame;
-    struct timeval tv;
-    int x, ret;
-
-    tv.tv_sec = tv.tv_usec = 0;
-    x = amqp_simple_wait_frame_noblock(state, &frame, &tv);
-
-    if (x != AMQP_STATUS_OK && x != AMQP_STATUS_TIMEOUT) {
-        logger(LOG_INFO,
-                "OpenLI mediator: unable to check status of an internal RMQ publishing socket");
-        return -1;
-    }
-
-    if (*is_blocked) {
-        ret = 0;
-    } else {
-        ret = 1;
-    }
-
-    if (x == AMQP_STATUS_TIMEOUT) {
-        return ret;
-    }
-
-    if (AMQP_FRAME_METHOD == frame.frame_type) {
-        switch(frame.payload.method.id) {
-            case AMQP_CONNECTION_BLOCKED_METHOD:
-                if ((*is_blocked) == 0) {
-                    logger(LOG_INFO,
-                            "OpenLI mediator: RMQ is unable to handle any more published ETSI records!");
-                    logger(LOG_INFO,
-                            "OpenLI mediator: this is a SERIOUS problem -- received ETSI records are going to be dropped!");
-                }
-                *is_blocked = 1;
-                ret = 0;
-                break;
-            case AMQP_CONNECTION_UNBLOCKED_METHOD:
-                if ((*is_blocked) == 1) {
-                    logger(LOG_INFO,
-                            "OpenLI mediator: RMQ has become unblocked and will resume publishing ETSI records.");
-                    ret = 0;
-                } else {
-                    ret = 1;
-                }
-                *is_blocked = 0;
-                break;
-            case AMQP_CONNECTION_CLOSE_METHOD:
-                logger(LOG_INFO,
-                        "OpenLI mediator: 'close' exception occurred on an internal RMQ connection -- must restart connection");
-                return -1;
-            case AMQP_CHANNEL_CLOSE_METHOD:
-                logger(LOG_INFO,
-                        "OpenLI mediator: channel exception occurred on an internal RMQ connection -- must reset connection");
-                return -1;
-        }
-    }
-    return ret;
-}
-
 /** Disables consumption from a RabbitMQ queue by an existing connection
  *
  *  @param state            The RMQ connection to disassociate the queue from
@@ -253,11 +189,6 @@ int declare_mediator_liid_RMQ_queue(amqp_connection_state_t state,
       return 0;
     }
 
-    /*
-    if (update_mediator_rmq_connection_block_status(state, is_blocked) < 0) {
-        return -1;
-    }
-    */
     if (*is_blocked) {
         return 0;
     }
@@ -285,11 +216,7 @@ int declare_mediator_rawip_RMQ_queue(amqp_connection_state_t state,
         char *liid, uint8_t *is_blocked) {
 
     char queuename[1024];
-    /*
-    if (update_mediator_rmq_connection_block_status(state, is_blocked) < 0) {
-        return -1;
-    }
-    */
+
     if (*is_blocked == 0) {
         snprintf(queuename, 1024, "%s-rawip", liid);
         return declare_RMQ_queue(state, queuename, 4);
@@ -332,10 +259,6 @@ static int produce_mediator_RMQ(amqp_connection_state_t state,
         props.expiration = amqp_cstring_bytes(expirystr);
     }
 
-    if (update_mediator_rmq_connection_block_status(state, is_blocked) < 0) {
-        return -1;
-    }
-
     if ((*is_blocked) == 0) {
         pub_ret = amqp_basic_publish(state, channel, amqp_cstring_bytes(""),
                 amqp_cstring_bytes(queuename), 0, 0, &props, message_bytes);
@@ -992,7 +915,7 @@ static int consume_other_frame(amqp_connection_state_t state) {
 int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
 
     size_t i;
-    int r, elapsed;
+    int r;
     amqp_frame_t frame;
     saved_received_data_t *sav;
     struct timeval tv;
@@ -1027,7 +950,6 @@ int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
         }
     }
 
-    elapsed = 0;
     while (cc_await > 0 || iri_await > 0 || raw_await > 0) {
         /* keep consuming until we get an ACK or a NACK -- everything else
          * can just be ignored (note that this will block until we get what
@@ -1038,13 +960,25 @@ int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
                 &tv);
 
         if (r == AMQP_STATUS_TIMEOUT) {
-            elapsed ++;
-            if (elapsed >= 3) {
-                /* we've gone 3 seconds with no acknowledgement, we'll
-                 * have to retry the publish
-                 */
-                return 0;
+            /* No acknowledgements this second, we need to move on instead so
+             * that we're not blocking and hope that something turns up
+             * next time we check for acks.
+             */
+            if (cc_await == 0) {
+                col->saved_cc_msg_cnt = 0;
+            }
+            if (iri_await == 0) {
+                col->saved_iri_msg_cnt = 0;
+            }
+            if (raw_await == 0) {
+                col->saved_raw_msg_cnt = 0;
             }
+            if (col->saved_raw_msg_cnt < MAX_SAVED_RECEIVED_DATA &&
+                    col->saved_iri_msg_cnt < MAX_SAVED_RECEIVED_DATA &&
+                    col->saved_cc_msg_cnt < MAX_SAVED_RECEIVED_DATA) {
+                col->queue_full = 0;
+            }
+            return 1;
         } else if (r != AMQP_STATUS_OK) {
             /* broker failure, must retry */
             return 0;
@@ -1055,7 +989,12 @@ int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
 
             switch(frame.payload.method.id) {
                 case AMQP_CONNECTION_CLOSE_METHOD:
+                    logger(LOG_INFO,
+                            "OpenLI mediator: 'close' exception occurred on an internal RMQ connection -- must restart connection");
+                    return 0;
                 case AMQP_CHANNEL_CLOSE_METHOD:
+                    logger(LOG_INFO,
+                            "OpenLI mediator: channel exception occurred on an internal RMQ connection -- must reset connection");
                     return 0;
                 case AMQP_BASIC_ACK_METHOD:
                     ack = (amqp_basic_ack_t *)frame.payload.method.decoded;
@@ -1078,7 +1017,6 @@ int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
                     } else {
                         break;
                     }
-
                     while (*start < MAX_SAVED_RECEIVED_DATA && *await > 0) {
                         saved_received_data_t *next = &(sav[*start]);
 
@@ -1103,6 +1041,22 @@ int consume_mediator_RMQ_producer_acks(coll_recv_t *col) {
 
                 case AMQP_BASIC_NACK_METHOD:
                     return 0;
+                case AMQP_CONNECTION_BLOCKED_METHOD:
+                    if ((col->rmq_blocked) == 0) {
+                        logger(LOG_INFO,
+                                "OpenLI mediator: RMQ is unable to handle any more published ETSI records!");
+                        logger(LOG_INFO,
+                                "OpenLI mediator: this is a SERIOUS problem -- received ETSI records are going to be dropped!");
+                    }
+                    col->rmq_blocked = 1;
+                    break;
+                case AMQP_CONNECTION_UNBLOCKED_METHOD:
+                    if ((col->rmq_blocked) == 1) {
+                        logger(LOG_INFO,
+                                "OpenLI mediator: RMQ has become unblocked and will resume publishing ETSI records.");
+                    }
+                    col->rmq_blocked = 0;
+                    break;
             }
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@`
`51`	`51`	`*`
`52`	`52`	`*/`
`53`	`53`
`54`		`-#define MAX_SAVED_RECEIVED_DATA 32`
	`54`	`+#define MAX_SAVED_RECEIVED_DATA 256`
`55`	`55`
`56`	`56`	`/** Types of messages that can be sent between the main mediator thread and a`
`57`	`57`	`* collector receive thread.`