From 335c7d54449458ed6553ee67bba93f19aa6c19d0 Mon Sep 17 00:00:00 2001
From: Satya Bodapati <satya.bodapati@percona.com>
Date: Fri, 4 Oct 2024 12:37:16 +0100
Subject: [PATCH] PXB-3374 : Space ID  is missing encryption information
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

With lock-ddl=REDUCED, the following sequence can happen (Not possible
with lock-ddl=ON)

    1. first an IBD with all zero(invalid) encryption is found
    2. so this is added to invalid encryption ids
    3. the same IBD is found again (because of concurrent DDL, they are both found with different names)
    4. this time IBD has proper encryption,  so fil_space_t is created
    5. encryption info from redo is parsed. fil_tablespace_redo_encryption()
    6. because the fil_space_t exists, encryption key is not added to the recovery encryption keys map
    7. later at the end of the backup, we check if we have found a valid encryption key for the invalid encryption space_ids.
    8. we haven’t (remember at step 6, we skipped it)
    9. backup is aborted

Fix:
----
Store keys in recv_sys->keys map even if there exists a tablespace
---
 storage/innobase/fil/fil0fil.cc | 43 ++++++++++++++++++---------------
 1 file changed, 23 insertions(+), 20 deletions(-)

diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 41f8e9c85ef..f9be354ee2f 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -11227,7 +11227,7 @@ const byte *fil_tablespace_redo_encryption(const byte *ptr, const byte *end,
         xb::error() << KEYRING_NOT_LOADED;
         exit(EXIT_FAILURE);
       }
-      return (ptr + len);
+      return (ptr);
     }
   } else {
     ulint master_key_id = mach_read_from_4(ptr + Encryption::MAGIC_SIZE);
@@ -11240,13 +11240,6 @@ const byte *fil_tablespace_redo_encryption(const byte *ptr, const byte *end,
 
   ut_ad(len == Encryption::INFO_SIZE);
 
-  if (space != nullptr) {
-    Encryption::set_or_generate(Encryption::AES, key, iv,
-                                space->m_encryption_metadata);
-    fsp_flags_set_encryption(space->flags);
-    return ptr;
-  }
-
   /* Space is not loaded yet. Remember this key in recv_sys and use it later
   to pupulate space encryption info once it is loaded. */
   DBUG_EXECUTE_IF("dont_update_key_found_during_REDO_scan", return ptr;);
@@ -11256,27 +11249,37 @@ const byte *fil_tablespace_redo_encryption(const byte *ptr, const byte *end,
         ut::new_withkey<recv_sys_t::Encryption_Keys>(UT_NEW_THIS_FILE_PSI_KEY);
   }
 
+  bool key_exists = false;
   /* Search if key entry already exists for this tablespace, update it. */
   for (auto &recv_key : *recv_sys->keys) {
     if (recv_key.space_id == space_id) {
       memcpy(recv_key.iv, iv, Encryption::KEY_LEN);
       memcpy(recv_key.ptr, key, Encryption::KEY_LEN);
       recv_key.lsn = lsn;
-      return ptr;
+      key_exists = true;
+      break;
     }
   }
 
-  /* No existing entry found, create new one and insert it. */
-  recv_sys_t::Encryption_Key new_key;
-  new_key.iv = static_cast<byte *>(
-      ut::malloc_withkey(UT_NEW_THIS_FILE_PSI_KEY, Encryption::KEY_LEN));
-  memcpy(new_key.iv, iv, Encryption::KEY_LEN);
-  new_key.ptr = static_cast<byte *>(
-      ut::malloc_withkey(UT_NEW_THIS_FILE_PSI_KEY, Encryption::KEY_LEN));
-  memcpy(new_key.ptr, key, Encryption::KEY_LEN);
-  new_key.space_id = space_id;
-  new_key.lsn = lsn;
-  recv_sys->keys->push_back(new_key);
+  if (!key_exists) {
+    /* No existing entry found, create new one and insert it. */
+    recv_sys_t::Encryption_Key new_key;
+    new_key.iv = static_cast<byte *>(
+        ut::malloc_withkey(UT_NEW_THIS_FILE_PSI_KEY, Encryption::KEY_LEN));
+    memcpy(new_key.iv, iv, Encryption::KEY_LEN);
+    new_key.ptr = static_cast<byte *>(
+        ut::malloc_withkey(UT_NEW_THIS_FILE_PSI_KEY, Encryption::KEY_LEN));
+    memcpy(new_key.ptr, key, Encryption::KEY_LEN);
+    new_key.space_id = space_id;
+    new_key.lsn = lsn;
+    recv_sys->keys->push_back(new_key);
+  }
+
+  if (space != nullptr) {
+    Encryption::set_or_generate(Encryption::AES, key, iv,
+                                space->m_encryption_metadata);
+    fsp_flags_set_encryption(space->flags);
+  }
 
   return ptr;
 }