Merge "res_pjsip_exten_state: Fix race condition between sending NOTIFY and termination"

4 files changed, 42 insertions, 7 deletions

diff --git a/include/asterisk/res_pjsip_pubsub.h b/include/asterisk/res_pjsip_pubsub.h
index d32b246af..afa0d6930 100644
--- a/include/asterisk/res_pjsip_pubsub.h
+++ b/include/asterisk/res_pjsip_pubsub.h
@@ -406,6 +406,16 @@ void ast_sip_subscription_get_remote_uri(struct ast_sip_subscription *sub, char
 const char *ast_sip_subscription_get_resource_name(struct ast_sip_subscription *sub);
 
 /*!
+ * \brief Get whether the subscription has been terminated or not.
+ *
+ * \param sub The subscription.
+ * \retval 0 not terminated.
+ * \retval 1 terminated.
+ * \since 13.4.0
+ */
+int ast_sip_subscription_is_terminated(const struct ast_sip_subscription *sub);
+
+/*!
  * \brief Get a header value for a subscription.
  *
  * For notifiers, the headers of the inbound SUBSCRIBE that started the dialog
diff --git a/res/res_pjsip_exten_state.c b/res/res_pjsip_exten_state.c
index da9b133f9..a05e1915d 100644
--- a/res/res_pjsip_exten_state.c
+++ b/res/res_pjsip_exten_state.c
@@ -37,6 +37,7 @@
 #include "asterisk/astobj2.h"
 #include "asterisk/sorcery.h"
 #include "asterisk/app.h"
+#include "asterisk/taskprocessor.h"
 
 #define BODY_SIZE 1024
 #define EVENT_TYPE_SIZE 50
@@ -53,6 +54,8 @@ struct exten_state_subscription {
 	int id;
 	/*! The SIP subscription */
 	struct ast_sip_subscription *sip_sub;
+	/*! The serializer to use for notifications */
+	struct ast_taskprocessor *serializer;
 	/*! Context in which subscription looks for updates */
 	char context[AST_MAX_CONTEXT];
 	/*! Extension within the context to receive updates from */
@@ -113,6 +116,7 @@ static void exten_state_subscription_destructor(void *obj)
 
 	ast_free(sub->user_agent);
 	ao2_cleanup(sub->sip_sub);
+	ast_taskprocessor_unreference(sub->serializer);
 }
 
 static char *get_user_agent(const struct ast_sip_subscription *sip_sub)
@@ -157,6 +161,13 @@ static struct exten_state_subscription *exten_state_subscription_alloc(
 	}
 
 	exten_state_sub->sip_sub = ao2_bump(sip_sub);
+
+	/* We keep our own reference to the serializer as there is no guarantee in state_changed
+	 * that the subscription tree is still valid when it is called. This can occur when
+	 * the subscription is terminated at around the same time as the state_changed
+	 * callback is invoked.
+	 */
+	exten_state_sub->serializer = ao2_bump(ast_sip_subscription_get_serializer(sip_sub));
 	exten_state_sub->last_exten_state = INITIAL_LAST_EXTEN_STATE;
 	exten_state_sub->last_presence_state = AST_PRESENCE_NOT_SET;
 	exten_state_sub->user_agent = get_user_agent(sip_sub);
@@ -205,11 +216,6 @@ static struct notify_task_data *alloc_notify_task_data(char *exten, struct exten
 	task_data->exten_state_data.device_state_info = ao2_bump(info->device_state_info);
 	task_data->exten_state_data.sub = exten_state_sub->sip_sub;
 
-	ast_sip_subscription_get_local_uri(exten_state_sub->sip_sub,
-			task_data->exten_state_data.local, sizeof(task_data->exten_state_data.local));
-	ast_sip_subscription_get_remote_uri(exten_state_sub->sip_sub,
-			task_data->exten_state_data.remote, sizeof(task_data->exten_state_data.remote));
-
 	if ((info->exten_state == AST_EXTENSION_DEACTIVATED) ||
 	    (info->exten_state == AST_EXTENSION_REMOVED)) {
 		ast_verb(2, "Watcher for hint %s %s\n", exten, info->exten_state
@@ -228,6 +234,19 @@ static int notify_task(void *obj)
 		.body_data = &task_data->exten_state_data,
 	};
 
+	/* Terminated subscriptions are no longer associated with a valid tree, and sending
+	 * NOTIFY messages on a subscription which has already been terminated won't work.
+	 */
+	if (ast_sip_subscription_is_terminated(task_data->exten_state_sub->sip_sub)) {
+		return 0;
+	}
+
+	/* All access to the subscription must occur within a task executed within its serializer */
+	ast_sip_subscription_get_local_uri(task_data->exten_state_sub->sip_sub,
+			task_data->exten_state_data.local, sizeof(task_data->exten_state_data.local));
+	ast_sip_subscription_get_remote_uri(task_data->exten_state_sub->sip_sub,
+			task_data->exten_state_data.remote, sizeof(task_data->exten_state_data.remote));
+
 	/* Pool allocation has to happen here so that we allocate within a PJLIB thread */
 	task_data->exten_state_data.pool = pjsip_endpt_create_pool(ast_sip_get_pjsip_endpoint(),
 			"exten_state", 1024, 1024);
@@ -263,8 +282,8 @@ static int state_changed(char *context, char *exten,
 
 	/* safe to push this async since we copy the data from info and
 	   add a ref for the device state info */
-	if (ast_sip_push_task(ast_sip_subscription_get_serializer(task_data->exten_state_sub->sip_sub),
-			      notify_task, task_data)) {
+	if (ast_sip_push_task(task_data->exten_state_sub->serializer, notify_task,
+		task_data)) {
 		ao2_cleanup(task_data);
 		return -1;
 	}
diff --git a/res/res_pjsip_pubsub.c b/res/res_pjsip_pubsub.c
index ebc43d15b..1ecb17d99 100644
--- a/res/res_pjsip_pubsub.c
+++ b/res/res_pjsip_pubsub.c
@@ -2185,6 +2185,11 @@ const char *ast_sip_subscription_get_resource_name(struct ast_sip_subscription *
 	return sub->resource;
 }
 
+int ast_sip_subscription_is_terminated(const struct ast_sip_subscription *sub)
+{
+	return sub->subscription_state == PJSIP_EVSUB_STATE_TERMINATED ? 1 : 0;
+}
+
 static int sip_subscription_accept(struct sip_subscription_tree *sub_tree, pjsip_rx_data *rdata, int response)
 {
 	pjsip_hdr res_hdr;
diff --git a/res/res_pjsip_pubsub.exports.in b/res/res_pjsip_pubsub.exports.in
index 2a6b75f00..58702d6c4 100644
--- a/res/res_pjsip_pubsub.exports.in
+++ b/res/res_pjsip_pubsub.exports.in
@@ -37,6 +37,7 @@
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_local_uri;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_remote_uri;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_header;
+		LINKER_SYMBOL_PREFIXast_sip_subscription_is_terminated;
 	local:
 		*;
 };