From 50ebe2f1d99e6fe1cc392b7dc6f51bb20e9d3a4d Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Wed, 26 Feb 2025 16:53:27 +0000 Subject: [PATCH] Add `retain_expiry_interval` option This fixes expired retained message not being removed from memory if they are not subscribed to. Closes #3221. --- ChangeLog.txt | 2 ++ man/mosquitto.conf.5.xml | 28 ++++++++++++++++++++++++++++ src/conf.c | 4 ++++ src/loop.c | 1 + src/mosquitto_broker_internal.h | 2 ++ src/retain.c | 31 +++++++++++++++++++++++++++++++ 6 files changed, 68 insertions(+) diff --git a/ChangeLog.txt b/ChangeLog.txt index 66de676b..03641151 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -8,6 +8,8 @@ Broker: an IPv6 link-local address and no other IPv6 addresses. Closes #2696. - Fix mismatched wrapped/unwrapped memory alloc/free in properties. Closes #3192. - Fix `allow_anonymous false` not being applied in local only mode. Closes #3198. +- Add `retain_expiry_interval` option to fix expired retained message not + being removed from memory if they are not subscribed to. Closes #3221. Client library: - Fix potential deadlock in mosquitto_sub if `-W` is used. Closes #3175. diff --git a/man/mosquitto.conf.5.xml b/man/mosquitto.conf.5.xml index 898d48be..16198949 100644 --- a/man/mosquitto.conf.5.xml +++ b/man/mosquitto.conf.5.xml @@ -987,6 +987,34 @@ log_timestamp_format %Y-%m-%dT%H:%M:%S Reloaded on reload signal. + + minutes + + + The default behaviour of mosquitto is to remove retained + messages that have reached their message-expiry-interval + property the next time that that message is accessed - + either by being replaced by a new message, or on the + next subscription that matches the message. If you have + a pattern of publishing many retained messages with a + message-expiry-interval, but that are not subscribed to, + then the expired retained messages will remain in + memory. This option configures the broker to + periodically check the retained tree for expired + messages. + + + + Defaults to off. Setting to a value greater than zero + means the broker will make a check at an interval of + that number of minutes. + + + This option applies globally. + + Reloaded on reload signal. + + [ true | false ] diff --git a/src/conf.c b/src/conf.c index f9433ea1..a62c4ca1 100644 --- a/src/conf.c +++ b/src/conf.c @@ -201,6 +201,7 @@ static void config__init_reload(struct mosquitto__config *config) config->persistent_client_expiration = 0; config->queue_qos0_messages = false; config->retain_available = true; + config->retain_expiry_interval = 0; config->set_tcp_nodelay = false; config->sys_interval = 10; config->upgrade_outgoing_qos = false; @@ -1917,6 +1918,9 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload, #endif }else if(!strcmp(token, "retain_available")){ if(conf__parse_bool(&token, token, &config->retain_available, saveptr)) return MOSQ_ERR_INVAL; + }else if(!strcmp(token, "retain_expiry_interval")){ + if(conf__parse_int(&token, token, &config->retain_expiry_interval, saveptr)) return MOSQ_ERR_INVAL; + config->retain_expiry_interval *= 60; }else if(!strcmp(token, "retry_interval")){ log__printf(NULL, MOSQ_LOG_WARNING, "Warning: The retry_interval option is no longer available."); }else if(!strcmp(token, "round_robin")){ diff --git a/src/loop.c b/src/loop.c index ec953398..0940bb77 100644 --- a/src/loop.c +++ b/src/loop.c @@ -187,6 +187,7 @@ int mosquitto_main_loop(struct mosquitto__listener_sock *listensock, int listens #endif while(run){ + retain__expire(); queue_plugin_msgs(); context__free_disused(); #ifdef WITH_SYS_TREE diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index aff6e948..4458434f 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -291,6 +291,7 @@ struct mosquitto__config { bool queue_qos0_messages; bool per_listener_settings; bool retain_available; + int retain_expiry_interval; bool set_tcp_nodelay; int sys_interval; bool upgrade_outgoing_qos; @@ -791,6 +792,7 @@ int retain__init(void); void retain__clean(struct mosquitto__retainhier **retainhier); int retain__queue(struct mosquitto *context, const char *sub, uint8_t sub_qos, uint32_t subscription_identifier); int retain__store(const char *topic, struct mosquitto_msg_store *stored, char **split_topics); +void retain__expire(void); /* ============================================================ * Security related functions diff --git a/src/retain.c b/src/retain.c index 6c303c96..45765e80 100644 --- a/src/retain.c +++ b/src/retain.c @@ -29,6 +29,8 @@ Contributors: #include "utlist.h" +static time_t next_expire_check = 0; + static struct mosquitto__retainhier *retain__add_hier_entry(struct mosquitto__retainhier *parent, struct mosquitto__retainhier **sibling, const char *topic, uint16_t len) { struct mosquitto__retainhier *child; @@ -320,3 +322,32 @@ void retain__clean(struct mosquitto__retainhier **retainhier) } } +static void retain__expire_search(struct mosquitto__retainhier *retainhier) +{ + struct mosquitto__retainhier *branch, *branch_tmp; + + HASH_ITER(hh, retainhier->children, branch, branch_tmp){ + if(branch->children){ + retain__expire_search(branch); + } + if(branch->retained){ + if(branch->retained->message_expiry_time > 0 && db.now_real_s >= branch->retained->message_expiry_time){ + db__msg_store_ref_dec(&branch->retained); + branch->retained = NULL; +#ifdef WITH_SYS_TREE + db.retained_count--; +#endif + retain__clean_empty_hierarchy(retainhier); + } + } + } +} + + +void retain__expire(void) +{ + if(db.config->retain_expiry_interval > 0 && db.now_s > next_expire_check){ + retain__expire_search(db.retains); + next_expire_check = db.now_s + db.config->retain_expiry_interval; + } +}