MDEV-24954 : 10.5.9 crashes on int wsrep::client_state::ordered_commit():...

MDEV-24954 : 10.5.9 crashes on int wsrep::client_state::ordered_commit(): Assertion `owning_thread_id_ == wsrep::this_thread::get_id()' failed.

Binlog group commit could lead to a situation where group commit leader
accesses participant thd's wsrep client state concurrently with the
thread executing the participant thd.

This is because of race condition in
MYSQL_BIN_LOG::write_transaction_to_binlog_events(),
and was fixed by moving wsrep_ordered_commit() to happen in
MYSQL_BIN_LOG::queue_for_group_commit() under protection of
LOCK_prepare_ordered mutex.

sql/log.cc

@@ -7471,6 +7471,8 @@ MYSQL_BIN_LOG::write_transaction_to_binlog(THD *thd,
   new transaction directly to participate in the group commit.
 
   @retval < 0   Error
+  @retval  -2   WSREP error with commit ordering
+  @retval  -3   WSREP return code to mark the leader
   @retval > 0   If queued as the first entry in the queue (meaning this
                 is the leader)
   @retval   0   Otherwise (queued as participant, leader handles the commit)
@@ -7768,6 +7770,22 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
     cur= entry->thd->wait_for_commit_ptr;
   }
 
+#ifdef WITH_WSREP
+  if (wsrep_is_active(entry->thd) &&
+      wsrep_run_commit_hook(entry->thd, entry->all))
+  {
+    /*  Release commit order here */
+    if (wsrep_ordered_commit(entry->thd, entry->all, wsrep_apply_error()))
+      result= -2;
+
+    /* return -3, if this is leader */
+    if (orig_queue == NULL)
+      result= -3;
+  }
+  else
+    DBUG_ASSERT(result != -2 && result != -3);
+#endif /* WITH_WSREP */
+
   if (opt_binlog_commit_wait_count > 0 && orig_queue != NULL)
     mysql_cond_signal(&COND_prepare_ordered);
   mysql_mutex_unlock(&LOCK_prepare_ordered);
@@ -7789,25 +7807,32 @@ MYSQL_BIN_LOG::write_transaction_to_binlog_events(group_commit_entry *entry)
 {
   int is_leader= queue_for_group_commit(entry);
 #ifdef WITH_WSREP
-  if (wsrep_is_active(entry->thd) &&
-      wsrep_run_commit_hook(entry->thd, entry->all))
-  {
-    /*
-      Release commit order and if leader, wait for prior commit to
-      complete. This establishes total order for group leaders.
-    */
-    if (wsrep_ordered_commit(entry->thd, entry->all, wsrep_apply_error()))
-    {
-      entry->thd->wakeup_subsequent_commits(1);
-      return 1;
-    }
-    if (is_leader)
-    {
-      if (entry->thd->wait_for_prior_commit())
-        return 1;
-    }
+  /* commit order was released in queue_for_group_commit() call,
+     here we check if wsrep_commit_ordered() failed or if we are leader */
+  switch (is_leader)
+  {
+  case -2: /* wsrep_ordered_commit() has failed */
+    DBUG_ASSERT(wsrep_is_active(entry->thd));
+    DBUG_ASSERT(wsrep_run_commit_hook(entry->thd, entry->all));
+    entry->thd->wakeup_subsequent_commits(1);
+    return true;
+  case -3: /* this is leader, wait for prior commit to
+              complete. This establishes total order for group leaders
+           */
+    DBUG_ASSERT(wsrep_is_active(entry->thd));
+    DBUG_ASSERT(wsrep_run_commit_hook(entry->thd, entry->all));
+    if (entry->thd->wait_for_prior_commit())
+      return true;
+
+    /* retain the correct is_leader value */
+    is_leader= 1;
+    break;
+
+  default: /* native MariaDB cases */
+    break;
   }
 #endif /* WITH_WSREP */
+
   /*
     The first in the queue handles group commit for all; the others just wait
     to be signalled when group commit is done.