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

@@ -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.

man/mosquitto.conf.5.xml

@@ -987,6 +987,34 @@ log_timestamp_format %Y-%m-%dT%H:%M:%S
 					<para>Reloaded on reload signal.</para>
 				</listitem>
 			</varlistentry>
+			<varlistentry>
+				<term><option>retain_expiry_interval</option> <replaceable>minutes</replaceable></term>
+				<listitem>
+					<para>
+						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.
+					</para>
+
+					<para>
+						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.
+					</para>
+
+					<para>This option applies globally.</para>
+
+					<para>Reloaded on reload signal.</para>
+				</listitem>
+			</varlistentry>
 			<varlistentry>
 				<term><option>set_tcp_nodelay</option> [ true | false ]</term>
 				<listitem>

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")){

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

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

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;
+	}
+}