From 23ac426608b658f1d73fda6f55e8d03ebf0660ec Mon Sep 17 00:00:00 2001
From: Erik Timan <erik.timan@emqx.io>
Date: Tue, 13 Dec 2022 09:14:04 +0100
Subject: [PATCH] fix(emqx_connector): check for key among prepared statements
 on query

An infinite loop was triggered in the mysql connector when a query
used a prepared statement key that was not among the defined prepared
statements on start. We now check that the key is defined among the
prepared statements before recursing. It seems that this bug was never
triggered in any production code flow and simply found while writing
tests.

An error return spell fix is also included as well as a FIXME comment
regarding running mysql:prepare and not distinguishing between
transient failures and syntax errors. Syntax errors should not be
retried.
---
 apps/emqx_connector/src/emqx_connector_mysql.erl | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl
index fc3068c66..b7cfd84b6 100644
--- a/apps/emqx_connector/src/emqx_connector_mysql.erl
+++ b/apps/emqx_connector/src/emqx_connector_mysql.erl
@@ -149,8 +149,12 @@ on_query(
     {SQLOrKey2, Data} = proc_sql_params(TypeOrKey, SQLOrKey, Params, State),
     case on_sql_query(InstId, MySqlFunction, SQLOrKey2, Data, Timeout, State) of
         {error, not_prepared} ->
-            case prepare_sql(Prepares, PoolName) of
+            case maybe_prepare_sql(SQLOrKey2, Prepares, PoolName) of
                 ok ->
+                    ?tp(
+                        mysql_connector_on_query_prepared_sql,
+                        #{type_or_key => TypeOrKey, sql_or_key => SQLOrKey, params => Params}
+                    ),
                     %% not return result, next loop will try again
                     on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, State);
                 {error, Reason} ->
@@ -182,7 +186,7 @@ on_batch_query(
         Request ->
             LogMeta = #{connector => InstId, first_request => Request, state => State},
             ?SLOG(error, LogMeta#{msg => "invalid request"}),
-            {error, invald_request}
+            {error, invalid_request}
     end.
 
 mysql_function(sql) ->
@@ -256,6 +260,12 @@ init_prepare(State = #{prepare_statement := Prepares, poolname := PoolName}) ->
             end
     end.
 
+maybe_prepare_sql(SQLOrKey, Prepares, PoolName) ->
+    case maps:is_key(SQLOrKey, Prepares) of
+        true -> prepare_sql(Prepares, PoolName);
+        false -> {error, prepared_statement_is_unprepared}
+    end.
+
 prepare_sql(Prepares, PoolName) when is_map(Prepares) ->
     prepare_sql(maps:to_list(Prepares), PoolName);
 prepare_sql(Prepares, PoolName) ->
@@ -305,6 +315,8 @@ prepare_sql_to_conn(Conn, [{Key, SQL} | PrepareList]) when is_pid(Conn) ->
             ?SLOG(info, LogMeta#{result => success}),
             prepare_sql_to_conn(Conn, PrepareList);
         {error, Reason} ->
+            % FIXME: we should try to differ on transient failers and
+            % syntax failures. Retrying syntax failures is not very productive.
             ?SLOG(error, LogMeta#{result => failed, reason => Reason}),
             {error, Reason}
     end.