MDEV-24449 Crash recovery may fail to apply some logbb-10.5-corruption

We are seeing !buf_pool.any_io_pending() assertion failures
in srv_start() ever since MDEV-21452 in 10.6. But, the
problem appears to be older.

recv_sys_t::apply(): At the end of each batch, wait for
pending reads to complete, so that any pending changes
will have been added to buf_pool.flush_list before we
flush the buffer pool.

io_callback(): Do not invoke read_slots->release()
before the callback function has returned, to ensure
the correct operation of recv_sys_t::apply().

3 files changed, 17 insertions, 5 deletions

diff --git a/storage/innobase/include/os0file.h b/storage/innobase/include/os0file.h
index ab1d5bb1dea..1e24803c8c1 100644
--- a/storage/innobase/include/os0file.h
+++ b/storage/innobase/include/os0file.h
@@ -1157,6 +1157,8 @@ be other, synchronous, pending writes. */
 void
 os_aio_wait_until_no_pending_writes();
 
+/** Wait until all pending asynchronous reads have completed. */
+void os_aio_wait_until_no_pending_reads();
 
 /** Prints info of the aio arrays.
 @param[in/out]	file		file where to print */
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index fe691a4b256..47d26ff3b7d 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -2698,6 +2698,7 @@ next_page:
     buf_pool.free_block(free_block);
 
     /* Wait until all the pages have been processed */
+
     while (!pages.empty())
     {
       const bool abort= found_corrupt_log || found_corrupt_fs;
@@ -2726,6 +2727,7 @@ next_page:
   mysql_mutex_assert_not_owner(&log_sys.mutex);
   mutex_exit(&mutex);
 
+  os_aio_wait_until_no_pending_reads();
   /* Instead of flushing, last_batch could sort the buf_pool.flush_list
   in ascending order of buf_page_t::oldest_modification. */
   buf_flush_sync();
diff --git a/storage/innobase/os/os0file.cc b/storage/innobase/os/os0file.cc
index e41809e4f1c..11a2b2ebb9b 100644
--- a/storage/innobase/os/os0file.cc
+++ b/storage/innobase/os/os0file.cc
@@ -3861,24 +3861,26 @@ os_file_get_status(
 
 extern void fil_aio_callback(const IORequest &request);
 
-static void io_callback(tpool::aiocb* cb)
+static void io_callback(tpool::aiocb *cb)
 {
   ut_a(cb->m_err == DB_SUCCESS);
-  const IORequest request(*static_cast<const IORequest*>
-                          (static_cast<const void*>(cb->m_userdata)));
+  const IORequest &request= *static_cast<const IORequest*>
+    (static_cast<const void*>(cb->m_userdata));
+
   /* Return cb back to cache*/
   if (cb->m_opcode == tpool::aio_opcode::AIO_PREAD)
   {
     ut_ad(read_slots->contains(cb));
+    fil_aio_callback(request);
     read_slots->release(cb);
   }
   else
   {
     ut_ad(write_slots->contains(cb));
+    const IORequest req{request};
     write_slots->release(cb);
+    fil_aio_callback(req);
   }
-
-  fil_aio_callback(request);
 }
 
 #ifdef LINUX_NATIVE_AIO
@@ -4074,6 +4076,12 @@ void os_aio_wait_until_no_pending_writes()
   buf_dblwr.wait_flush_buffered_writes();
 }
 
+/** Wait until all pending asynchronous reads have completed. */
+void os_aio_wait_until_no_pending_reads()
+{
+  read_slots->wait();
+}
+
 /** Request a read or write.
 @param type		I/O request
 @param buf		buffer