res_pjsip_mwi: add better handling of solicited vs unsolicited subscriptions

res_pjsip_mwi allows both solicited and unsolicited MWI subscription types.
While both can be set in the configuration for a given endpoint/aor, only
one is allowed. Precedence is given to unsolicited. Meaning if an endpoint/aor
is configured to allow both types then the solicited subscription is rejected
when it comes in. However, there is a configuration option to override that
behavior:

mwi_subscribe_replaces_unsolicited

When set to "yes" then when a solicited subscription comes in instead of
rejecting it Asterisk is suppose to replace the unsolicited one if it exists.
Prior to this patch there was a bug in Asterisk that allowed the solicted one
to be added, but did not remove the unsolicited. As a matter of fact a new
unsolicited subscription got added everytime a SIP register was received.
Over time this eventually could "flood" a phone with SIP notifies.

This patch fixes that behavior to now make it work as expected. If configured
to do so a solicited subscription now properly replaces the unsolicited one.
As well when an unsubscribe is received the unsolicited subscription is
restored. Logic was also put in to handle reloads, and any configuration changes
that might result from that. For instance, if a solicited subscription had
previously replaced an unsolicited one, but after reload it was configured to
not allow that then the solicited one needs to be shutdown, and the unsolicited
one added.

ASTERISK-28488

Change-Id: Iec2ec12d9431097e97ed5f37119963aee41af7b1

res/res_pjsip_mwi.c

@@ -42,6 +42,7 @@
 
 struct mwi_subscription;
 static struct ao2_container *unsolicited_mwi;
+static struct ao2_container *solicited_mwi;
 
 static char *default_voicemail_extension;
 
@@ -119,6 +120,8 @@ struct mwi_subscription {
 	char *aors;
 	/*! Is the MWI solicited (i.e. Initiated with an external SUBSCRIBE) ? */
 	unsigned int is_solicited;
+	/*! True if this subscription is to be terminated */
+	unsigned int terminate;
 	/*! Identifier for the subscription.
 	 * The identifier is the same as the corresponding endpoint's stasis ID.
 	 * Used as a hash key
@@ -665,7 +668,7 @@ static void send_mwi_notify(struct mwi_subscription *sub)
 
 		ao2_cleanup(aor);
 		ao2_cleanup(endpoint);
-		ast_sip_subscription_notify(sub->sip_sub, &data, 0);
+		ast_sip_subscription_notify(sub->sip_sub, &data, sub->terminate);
 
 		return;
 	}
@@ -676,18 +679,22 @@ static void send_mwi_notify(struct mwi_subscription *sub)
 static int unsubscribe_stasis(void *obj, void *arg, int flags)
 {
 	struct mwi_stasis_subscription *mwi_stasis = obj;
+
 	if (mwi_stasis->mwi_subscriber) {
 		ast_debug(3, "Removing stasis subscription to mailbox %s\n", mwi_stasis->mailbox);
 		mwi_stasis->mwi_subscriber = ast_mwi_unsubscribe_and_join(mwi_stasis->mwi_subscriber);
 	}
-
 	return CMP_MATCH;
 }
 
+static int create_unsolicited_mwi_subscriptions(struct ast_sip_endpoint *endpoint,
+		int recreate, int send_now);
+
 static void mwi_subscription_shutdown(struct ast_sip_subscription *sub)
 {
 	struct mwi_subscription *mwi_sub;
 	struct ast_datastore *mwi_datastore;
+	struct ast_sip_endpoint *endpoint = NULL;
 
 	mwi_datastore = ast_sip_subscription_get_datastore(sub, MWI_DATASTORE);
 	if (!mwi_datastore) {
@@ -695,10 +702,25 @@ static void mwi_subscription_shutdown(struct ast_sip_subscription *sub)
 	}
 
 	mwi_sub = mwi_datastore->data;
+
 	ao2_callback(mwi_sub->stasis_subs, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe_stasis, NULL);
 	ast_sip_subscription_remove_datastore(sub, MWI_DATASTORE);
+	endpoint = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", mwi_sub->id);
 
 	ao2_ref(mwi_datastore, -1);
+	ao2_unlink(solicited_mwi, mwi_sub);
+
+	/*
+	 * When a solicited subscription is removed it's possible an unsolicited one
+	 * needs to be [re-]created. Attempt to establish unsolicited MWI.
+	 */
+	if (unsolicited_mwi && endpoint) {
+		ao2_lock(unsolicited_mwi);
+		create_unsolicited_mwi_subscriptions(endpoint, 1, 1);
+		ao2_unlock(unsolicited_mwi);
+	}
+
+	ao2_cleanup(endpoint);
 }
 
 static void mwi_ds_destroy(void *data)
@@ -734,43 +756,165 @@ static int add_mwi_datastore(struct mwi_subscription *sub)
 }
 
 /*!
- * \brief Determines if an endpoint is receiving unsolicited MWI for a particular mailbox.
+ * \internal
+ * \brief Determine if an MWI subscription already exists for the given endpoint/mailbox
+ *
+ * Search the given container, and attempt to find out if the given endpoint has a
+ * current subscription within. If so pass back the associated mwi_subscription and
+ * mwi_stasis_subscription objects.
+ *
+ * \note If a subscription is located then the caller is responsible for removing the
+ * references to the passed back mwi_subscription and mwi_stasis_subscription objects.
+ *
+ * \note Must be called with the given container already locked.
  *
- * \param endpoint The endpoint to check
- * \param mailbox The candidate mailbox
- * \retval 0 The endpoint does not receive unsolicited MWI for this mailbox
- * \retval 1 The endpoint receives unsolicited MWI for this mailbox
+ * \param container The ao2_container to search
+ * \param endpoint The endpoint to find
+ * \param mailbox The mailbox potentially subscribed
+ * \param mwi_sub [out] May contain the located mwi_subscription
+ * \param mwi_stasis [out] May contain the located mwi_stasis_subscription
+ *
+ * \retval 1 if a subscription was located, 0 otherwise
  */
-static int endpoint_receives_unsolicited_mwi_for_mailbox(struct ast_sip_endpoint *endpoint,
-		const char *mailbox)
+static int has_mwi_subscription(struct ao2_container *container,
+		struct ast_sip_endpoint *endpoint, const char *mailbox,
+		struct mwi_subscription **mwi_sub, struct mwi_stasis_subscription **mwi_stasis)
 {
 	struct ao2_iterator *mwi_subs;
-	struct mwi_subscription *mwi_sub;
-	const char *endpoint_id = ast_sorcery_object_get_id(endpoint);
-	int ret = 0;
 
-	mwi_subs = ao2_find(unsolicited_mwi, endpoint_id, OBJ_SEARCH_KEY | OBJ_MULTIPLE);
+	*mwi_sub = NULL;
+	*mwi_stasis = NULL;
+
+	mwi_subs = ao2_find(container, ast_sorcery_object_get_id(endpoint),
+						OBJ_SEARCH_KEY | OBJ_MULTIPLE | OBJ_NOLOCK);
 	if (!mwi_subs) {
 		return 0;
 	}
 
-	for (; (mwi_sub = ao2_iterator_next(mwi_subs)) && !ret; ao2_cleanup(mwi_sub)) {
-		struct mwi_stasis_subscription *mwi_stasis;
-
-		mwi_stasis = ao2_find(mwi_sub->stasis_subs, mailbox, OBJ_SEARCH_KEY);
-		if (mwi_stasis) {
-			if (endpoint->subscription.mwi.subscribe_replaces_unsolicited) {
-				unsubscribe_stasis(mwi_stasis, NULL, 0);
-				ao2_unlink(mwi_sub->stasis_subs, mwi_stasis);
-			} else {
-				ret = 1;
-			}
-			ao2_cleanup(mwi_stasis);
+	while ((*mwi_sub = ao2_iterator_next(mwi_subs))) {
+		*mwi_stasis = ao2_find((*mwi_sub)->stasis_subs, mailbox, OBJ_SEARCH_KEY);
+		if (*mwi_stasis) {
+			/* If found then caller is responsible for unrefs of passed back objects */
+			break;
 		}
+		ao2_ref(*mwi_sub, -1);
 	}
 
 	ao2_iterator_destroy(mwi_subs);
-	return ret;
+	return *mwi_stasis ? 1 : 0;
+}
+
+/*!
+ * \internal
+ * \brief Allow and/or replace the unsolicited subscription
+ *
+ * Checks to see if solicited subscription is allowed. If allowed, and an
+ * unsolicited one exists then prepare for replacement by removing the
+ * current unsolicited subscription.
+ *
+ * \param endpoint The endpoint
+ * \param mailbox The mailbox
+ *
+ * \retval 1 if a solicited subscription is allowed for the endpoint/mailbox
+ *         0 otherwise
+ */
+static int allow_and_or_replace_unsolicited(struct ast_sip_endpoint *endpoint, const char *mailbox)
+{
+	struct mwi_subscription *mwi_sub;
+	struct mwi_stasis_subscription *mwi_stasis;
+
+	if (!has_mwi_subscription(unsolicited_mwi, endpoint, mailbox, &mwi_sub, &mwi_stasis)) {
+		/* If no unsolicited subscription then allow the solicited one */
+		return 1;
+	}
+
+	if (!endpoint->subscription.mwi.subscribe_replaces_unsolicited) {
+		/* Has unsolicited subscription and can't replace, so disallow */
+		ao2_ref(mwi_stasis, -1);
+		ao2_ref(mwi_sub, -1);
+		return 0;
+	}
+
+	/*
+	 * The unsolicited subscription exists, and it is allowed to be replaced.
+	 * So, first remove the unsolicited stasis subscription, and if aggregation
+	 * is not enabled then also remove the mwi_subscription object as well.
+	 */
+	ast_debug(1, "Unsolicited subscription being replaced by solicited for "
+			"endpoint '%s' mailbox '%s'\n", ast_sorcery_object_get_id(endpoint), mailbox);
+
+	unsubscribe_stasis(mwi_stasis, NULL, 0);
+	ao2_unlink(mwi_sub->stasis_subs, mwi_stasis);
+
+	if (!endpoint->subscription.mwi.aggregate) {
+		ao2_unlink(unsolicited_mwi, mwi_sub);
+	}
+
+	ao2_ref(mwi_stasis, -1);
+	ao2_ref(mwi_sub, -1);
+
+	/* This solicited subscription is replacing an unsolicited one, so allow */
+	return 1;
+}
+
+static int send_notify(void *obj, void *arg, int flags);
+
+/*!
+ * \internal
+ * \brief Determine if an unsolicited MWI subscription is allowed
+ *
+ * \param endpoint The endpoint
+ * \param mailbox The mailbox
+ *
+ * \retval 1 if an unsolicited subscription is allowed for the endpoint/mailbox
+ *         0 otherwise
+ */
+static int is_unsolicited_allowed(struct ast_sip_endpoint *endpoint, const char *mailbox)
+{
+	struct mwi_subscription *mwi_sub;
+	struct mwi_stasis_subscription *mwi_stasis;
+
+	if (ast_strlen_zero(mailbox)) {
+		return 0;
+	}
+
+	/*
+	 * First check if an unsolicited subscription exists. If it does then we don't
+	 * want to add another one.
+	 */
+	if (has_mwi_subscription(unsolicited_mwi, endpoint, mailbox, &mwi_sub, &mwi_stasis)) {
+		ao2_ref(mwi_stasis, -1);
+		ao2_ref(mwi_sub, -1);
+		return 0;
+	}
+
+	/*
+	 * If there is no unsolicited subscription, next check to see if a solicited
+	 * subscription exists for the endpoint/mailbox. If not, then allow.
+	 */
+	if (!has_mwi_subscription(solicited_mwi, endpoint, mailbox, &mwi_sub, &mwi_stasis)) {
+		return 1;
+	}
+
+	/*
+	 * If however, a solicited subscription does exist then we'll need to see if that
+	 * subscription is allowed to replace the unsolicited one. If is allowed to replace
+	 * then disallow the unsolicited one.
+	 */
+	if (endpoint->subscription.mwi.subscribe_replaces_unsolicited) {
+		ao2_ref(mwi_stasis, -1);
+		ao2_ref(mwi_sub, -1);
+		return 0;
+	}
+
+	/* Otherwise, shutdown the solicited subscription and allow the unsolicited */
+	mwi_sub->terminate = 1;
+	send_notify(mwi_sub, NULL, 0);
+
+	ao2_ref(mwi_stasis, -1);
+	ao2_ref(mwi_sub, -1);
+
+	return 1;
 }
 
 /*!
@@ -796,19 +940,23 @@ static int mwi_validate_for_aor(void *obj, void *arg, int flags)
 		return 0;
 	}
 
+	/* A reload could be taking place so lock while checking if allowed */
+	ao2_lock(unsolicited_mwi);
 	mailboxes = ast_strdupa(aor->mailboxes);
 	while ((mailbox = ast_strip(strsep(&mailboxes, ",")))) {
 		if (ast_strlen_zero(mailbox)) {
 			continue;
 		}
 
-		if (endpoint_receives_unsolicited_mwi_for_mailbox(endpoint, mailbox)) {
+		if (!allow_and_or_replace_unsolicited(endpoint, mailbox)) {
 			ast_debug(1, "Endpoint '%s' already configured for unsolicited MWI for mailbox '%s'. "
 					"Denying MWI subscription to %s\n", ast_sorcery_object_get_id(endpoint), mailbox,
 					ast_sorcery_object_get_id(aor));
+			ao2_unlock(unsolicited_mwi);
 			return -1;
 		}
 	}
+	ao2_unlock(unsolicited_mwi);
 
 	return 0;
 }
@@ -954,6 +1102,7 @@ static int mwi_subscription_established(struct ast_sip_subscription *sip_sub)
 		ast_sip_subscription_remove_datastore(sip_sub, MWI_DATASTORE);
 	}
 
+	ao2_link(solicited_mwi, sub);
 	ao2_cleanup(sub);
 	ao2_cleanup(endpoint);
 	return 0;
@@ -1090,12 +1239,25 @@ static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub,
 	}
 }
 
-/*! \note Called with the unsolicited_mwi container lock held. */
-static int create_mwi_subscriptions_for_endpoint(void *obj, void *arg, int flags)
+/*!
+ * \internal
+ * \brief Create unsolicited MWI subscriptions for an endpoint
+ *
+ * \note Call with the unsolicited_mwi container lock held.
+ *
+ * \param endpoint An endpoint object
+ * \param recreate Whether or not unsolicited subscriptions are potentially being recreated
+ * \param send_now Whether or not to send a notify once the subscription is created
+ *
+ * \retval 0
+ */
+static int create_unsolicited_mwi_subscriptions(struct ast_sip_endpoint *endpoint,
+		int recreate, int send_now)
 {
 	RAII_VAR(struct mwi_subscription *, aggregate_sub, NULL, ao2_cleanup);
-	struct ast_sip_endpoint *endpoint = obj;
-	char *mailboxes, *mailbox;
+	char *mailboxes;
+	char *mailbox;
+	int sub_added = 0;
 
 	if (ast_strlen_zero(endpoint->subscription.mwi.mailboxes)) {
 		return 0;
@@ -1104,45 +1266,83 @@ static int create_mwi_subscriptions_for_endpoint(void *obj, void *arg, int flags
 	if (endpoint->subscription.mwi.aggregate) {
 		const char *endpoint_id = ast_sorcery_object_get_id(endpoint);
 
-		/* Check if subscription exists */
+		/* Check if aggregate subscription exists */
 		aggregate_sub = ao2_find(unsolicited_mwi, endpoint_id, OBJ_SEARCH_KEY | OBJ_NOLOCK);
-		if (aggregate_sub) {
+
+		/*
+		 * If enabled there should only ever exist a single aggregate subscription object
+		 * for an endpoint. So if it exists just return unless subscriptions are potentially
+		 * being added back in. If that's the case then continue.
+		 */
+		if (aggregate_sub && !recreate) {
 			return 0;
 		}
-		aggregate_sub = mwi_subscription_alloc(endpoint, 0, NULL);
+
 		if (!aggregate_sub) {
-			return 0;
+			aggregate_sub = mwi_subscription_alloc(endpoint, 0, NULL);
+			if (!aggregate_sub) {
+				return 0; /* No MWI aggregation for you */
+			}
 		}
 	}
 
+	/* Lock solicited so we don't potentially add to both containers */
+	ao2_lock(solicited_mwi);
+
 	mailboxes = ast_strdupa(endpoint->subscription.mwi.mailboxes);
 	while ((mailbox = ast_strip(strsep(&mailboxes, ",")))) {
 		struct mwi_subscription *sub;
 		struct mwi_stasis_subscription *mwi_stasis_sub;
 
-		/* check if subscription exists */
-		if (ast_strlen_zero(mailbox) ||
-			(!aggregate_sub && endpoint_receives_unsolicited_mwi_for_mailbox(endpoint, mailbox))) {
+		if (!is_unsolicited_allowed(endpoint, mailbox)) {
 			continue;
 		}
 
 		sub = aggregate_sub ?: mwi_subscription_alloc(endpoint, 0, NULL);
+		if (!sub) {
+			continue;
+		}
+
 		mwi_stasis_sub = mwi_stasis_subscription_alloc(mailbox, sub);
 		if (mwi_stasis_sub) {
 			ao2_link(sub->stasis_subs, mwi_stasis_sub);
 			ao2_ref(mwi_stasis_sub, -1);
 		}
-		if (!aggregate_sub && sub) {
+		if (!aggregate_sub) {
 			ao2_link_flags(unsolicited_mwi, sub, OBJ_NOLOCK);
+			if (send_now) {
+				send_notify(sub, NULL, 0);
+			}
 			ao2_ref(sub, -1);
 		}
+
+		if (aggregate_sub && !sub_added) {
+			/* If aggregation track if at least one subscription has been added */
+			sub_added = 1;
+		}
 	}
+
 	if (aggregate_sub) {
-		ao2_link_flags(unsolicited_mwi, aggregate_sub, OBJ_NOLOCK);
+		if (ao2_container_count(aggregate_sub->stasis_subs)) {
+			ao2_link_flags(unsolicited_mwi, aggregate_sub, OBJ_NOLOCK);
+			if (send_now && sub_added) {
+				send_notify(aggregate_sub, NULL, 0);
+			}
+		} else {
+			/* No stasis subscriptions then no MWI data to aggregate */
+			ao2_ref(aggregate_sub, -1);
+		}
 	}
+
+	ao2_unlock(solicited_mwi);
 	return 0;
 }
 
+static int create_mwi_subscriptions_for_endpoint(void *obj, void *arg, int flags)
+{
+	return create_unsolicited_mwi_subscriptions(obj, 0, 0);
+}
+
 static int unsubscribe(void *obj, void *arg, int flags)
 {
 	struct mwi_subscription *mwi_sub = obj;
@@ -1347,11 +1547,20 @@ static int load_module(void)
 		ast_log(AST_LOG_WARNING, "Failed to create MWI serializer pool. The default SIP pool will be used for MWI\n");
 	}
 
+	solicited_mwi = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, MWI_BUCKETS,
+		mwi_sub_hash, NULL, mwi_sub_cmp);
+	if (!solicited_mwi) {
+		mwi_serializer_pool_shutdown();
+		ast_sip_unregister_subscription_handler(&mwi_handler);
+		return AST_MODULE_LOAD_DECLINE;
+	}
+
 	unsolicited_mwi = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0, MWI_BUCKETS,
 		mwi_sub_hash, NULL, mwi_sub_cmp);
 	if (!unsolicited_mwi) {
 		mwi_serializer_pool_shutdown();
 		ast_sip_unregister_subscription_handler(&mwi_handler);
+		ao2_ref(solicited_mwi, -1);
 		return AST_MODULE_LOAD_DECLINE;
 	}
 
@@ -1384,6 +1593,8 @@ static int unload_module(void)
 	ao2_ref(unsolicited_mwi, -1);
 	unsolicited_mwi = NULL;
 
+	ao2_cleanup(solicited_mwi);
+
 	mwi_serializer_pool_shutdown();
 
 	ast_sip_unregister_subscription_handler(&mwi_handler);