MDEV-36024: Redesign innodb_encrypt_log=ON

The innodb_encrypt_log=ON subformat of FORMAT_10_8 is inefficient,
because a new encryption or decryption context is being set up for
every log record payload snippet.

An in-place conversion between the old and new innodb_encrypt_log=ON
format is technically possible. No such conversion has been
implemented, though. There is some overhead with respect to the
unencrypted format (innodb_encrypt_log=OFF): At the end of each
mini-transaction, right before the CRC-32C, additional 8 bytes will be
reserved for a nonce (really, log_sys.get_flushed_lsn()), which forms
a part of an initialization vector.

log_t::FORMAT_ENC_11: The new format identifier, a UTF-8 encoding of
🗝 U+1F5DD OLD KEY (encryption). In this format, everything except the
types and lengths of log records will be encrypted. Thus, unlike in
FORMAT_10_8, also page identifiers and FILE_ records will be encrypted.
The initialization vector (IV) consists of the 8-byte nonce as well as
the type and length byte(s) of the first record of the mini-transaction.
Page identifiers will no longer form any part of the IV.

The old log_t::FORMAT_ENC_10_8 (innodb_encrypt_log=ON) will be supported
both by mariadb-backup and by crash recovery. Downgrade from the new
format will only be possible if the new server has been running or
restarted with innodb_encrypt_log=OFF. If innodb_encrypt_log=ON,
only the new log_t::FORMAT_ENC_11 will be written.

log_t::is_recoverable(): A new predicate, which holds for all 3
formats.

recv_sys_t::tmp_buf: A heap-allocated buffer for decrypting a
mini-transaction, or for making the wrap-around of a memory-mapped
log file contiguous.

recv_sys_t::start_lsn: The start of the mini-transaction.
Updated at the start of parse_tail().

log_decrypt_mtr(): Decrypt a mini-transaction in recv_sys.tmp_buf.
Theoretically, when reading the log via pread() rather than a read-only
memory mapping, we could modify the contents of log_sys.buf in place.
If we did that, we would have to re-read the last log block into
log_sys.buf before resuming writes, because otherwise that block could be
re-written as a mix of old decrypted data and new encrypted data, which
would cause a subsequent recovery failure unless the log checkpoint had
been advanced beyond this point.

log_decrypt_legacy(): Decrypt a log_t::FORMAT_ENC_10_8 record snippet
on stack. Replaces recv_buf::copy_if_needed().

recv_sys_t::get_backup_parser(): Return a recv_sys_t::parser, that is,
a pointer to an instantiation of parse_mmap or parse_mtr for the current
log format.

recv_sys_t::parse_mtr(), recv_sys_t::parse_mmap(): Add a parameter
template<uint32_t> for the current log_sys.format.

log_parse_start(): Validate the CRC-32C of a mini-transaction.
This has been split from the recv_sys_t::parse() template to
reduce code duplication. These two are the lowest-level functions
that will be instantiated for both recv_buf and recv_ring.

recv_sys_t::parse(): Split into ::log_parse_start() and parse_tail().
Add a parameter template<uint32_t format> to specialize for
log_sys.format at compilation time.

recv_sys_t::parse_tail(): Operate on pointers to contiguous
mini-transaction data. Use a parameter template<bool ENC_10_8>
for special handling of the old innodb_encrypt_log=ON format.
The former recv_buf::get_buf() is being inlined here.
Much of the logic is split into non-inline functions, to avoid
duplicating a lot of code for every template expansion.

log_crypt: Encrypt or decrypt a mini-transaction in place in the
new innodb_encrypt_log=ON format. We will use temporary buffers
so that encryption_ctx_update() can be invoked on integer multiples
of MY_AES_BLOCK_SIZE, except for the last bytes of the encrypted
payload, which will be encrypted or decrypted in place thanks to
ENCRYPTION_FLAG_NOPAD.

log_crypt::append(): Invoke encryption_ctx_update() in MY_AES_BLOCK_SIZE
(16-byte) blocks and scatter/gather shorter data blocks as needed.

log_crypt::finish(), Handle the last (possibly incomplete) block as a
special case, with ENCRYPTION_FLAG_NOPAD.

mtr_t::parse_length(): Parse the length of a log record.

mtr_t::encrypt(): Use log_crypt instead of the old log_encrypt_buf().

recv_buf::crc32c(): Add a parameter for the initial CRC-32C value.

recv_sys_t::rewind(): Operate on pointers to the start of the
mini-transaction and to the first skipped record.

recv_sys_t::trim(): Declare as ATTRIBUTE_COLD so that this rarely
invoked function will not be expanded inline in parse_tail().

recv_sys_t::parse_init(): Handle INIT_PAGE or FREE_PAGE while scanning
to the end of the log.

recv_sys_t::parse_page0(): Handle WRITE to FSP_SPACE_SIZE and
FSP_SPACE_FLAGS.

recv_sys_t::parse_store_if_exists(), recv_sys_t::parse_store(),
recv_sys_t::parse_oom(): Handle page-level log records.

mlog_decode_varint_length(): Make use of __builtin_clz() to avoid a loop
when possible.

mlog_decode_varint(): Define only on const byte*, as
ATTRIBUTE_NOINLINE static because it is a rather large function.

recv_buf::decode_varint(): Trivial wrapper for mlog_decode_varint().

recv_ring::decode_varint(): Special implementation.

log_page_modify(): Note that a page will be modified in recovery.
Split from recv_sys_t::parse_tail().

log_parse_file(): Handle non-page log records.

log_record_corrupted(), log_unknown(), log_page_id_corrupted():
Common error reporting functions.

Author: dr-m

extra/mariabackup/xtrabackup.cc

@@ -203,6 +203,8 @@ struct xb_filter_entry_t{
 
 /** whether log_copying_thread() is active; protected by recv_sys.mutex */
 static bool log_copying_running;
+/** the log parsing function for --backup */
+static recv_sys_t::parser backup_log_parse;
 /** for --backup, target LSN to copy the log to; protected by recv_sys.mutex */
 lsn_t metadata_to_lsn;
 
@@ -2640,7 +2642,10 @@ static byte log_hdr_buf[log_t::START_OFFSET + SIZE_OF_FILE_CHECKPOINT];
 static void log_hdr_init()
 {
   memset(log_hdr_buf, 0, sizeof log_hdr_buf);
-  mach_write_to_4(LOG_HEADER_FORMAT + log_hdr_buf, log_t::FORMAT_10_8);
+  /* log_t::FORMAT_ENC_10_8 is written to the file as FORMAT_10_8 */
+  mach_write_to_4(LOG_HEADER_FORMAT + log_hdr_buf,
+                  log_sys.format == log_t::FORMAT_ENC_11
+                  ? log_t::FORMAT_ENC_11 : log_t::FORMAT_10_8);
   mach_write_to_8(LOG_HEADER_START_LSN + log_hdr_buf,
                   log_sys.next_checkpoint_lsn);
   snprintf(reinterpret_cast<char*>(LOG_HEADER_CREATOR + log_hdr_buf),
@@ -3441,8 +3446,7 @@ static bool xtrabackup_copy_mmap_logfile()
   const byte *start= &log_sys.buf[recv_sys.offset];
   ut_d(recv_sys_t::parse_mtr_result r);
 
-  if ((ut_d(r=) recv_sys.parse_mmap<recv_sys_t::store::BACKUP>(false)) ==
-      recv_sys_t::OK)
+  if ((ut_d(r=) backup_log_parse(false)) == recv_sys_t::OK)
   {
     do
     {
@@ -3460,8 +3464,7 @@ static bool xtrabackup_copy_mmap_logfile()
         start = seq + 1;
       }
     }
-    while ((ut_d(r=) recv_sys.parse_mmap<recv_sys_t::store::BACKUP>(false)) ==
-           recv_sys_t::OK);
+    while ((ut_d(r=) backup_log_parse(false)) == recv_sys_t::OK);
 
     if (xtrabackup_copy_mmap_snippet(dst_log_file, start,
                                      &log_sys.buf[recv_sys.offset]))
@@ -3534,8 +3537,7 @@ static bool xtrabackup_copy_logfile(bool early_exit)
       if (log_sys.buf[recv_sys.offset] <= 1)
         break;
 
-      if (recv_sys.parse_mtr<recv_sys_t::store::BACKUP>(false) ==
-          recv_sys_t::OK)
+      if (backup_log_parse(false) == recv_sys_t::OK)
       {
         do
         {
@@ -3545,8 +3547,7 @@ static bool xtrabackup_copy_logfile(bool early_exit)
                                                  sequence_offset));
           *seq= 1;
         }
-        while ((r= recv_sys.parse_mtr<recv_sys_t::store::BACKUP>(false)) ==
-               recv_sys_t::OK);
+        while ((r= backup_log_parse(false)) == recv_sys_t::OK);
 
         if (ds_write(dst_log_file, log_sys.buf + start_offset,
                      recv_sys.offset - start_offset))
@@ -5585,6 +5586,7 @@ static bool xtrabackup_backup_func()
 	/* copy log file by current position */
 
 	mysql_mutex_lock(&recv_sys.mutex);
+	backup_log_parse = recv_sys.get_backup_parser();
 	recv_sys.lsn = log_sys.next_checkpoint_lsn;
 
 	const bool log_copy_failed = xtrabackup_copy_logfile(true);

mysql-test/suite/encryption/t/recovery_memory.test

@@ -43,3 +43,5 @@ let $restart_parameters=;
 SELECT COUNT(*) FROM t1;
 ALTER TABLE t1 FORCE;
 DROP TABLE t1;
+
+--rmdir $basedir

storage/innobase/handler/ha_innodb.cc

@@ -3681,6 +3681,8 @@ static int innodb_init_abort()
 {
 	DBUG_ENTER("innodb_init_abort");
 
+	recv_sys.tmp_free();
+
 	if (fil_system.temp_space) {
 		fil_system.temp_space->close();
 	}

storage/innobase/include/log0crypt.h

@@ -79,11 +79,10 @@ ATTRIBUTE_COLD bool log_decrypt(byte* buf, lsn_t lsn, ulint size);
 @return buf */
 byte *log_decrypt_buf(const byte *iv, byte *buf, const byte *data, uint len);
 
-/** Decrypt a log snippet.
-@param iv    initialization vector
-@param buf   buffer to be replaced with encrypted contents
-@param end   pointer past the end of buf */
-void log_decrypt_buf(const byte *iv, byte *buf, const byte *const end);
+/** Decrypt a mini-transaction in place.
+@param buf   start of the mini-transaction
+@param end   end of data (followed by sequence byte and the 8-byte nonce) */
+void log_decrypt_mtr(byte *buf, const byte *end) noexcept;
 
 /** Encrypt or decrypt a temporary file block.
 @param[in]	src		block to encrypt or decrypt

storage/innobase/include/log0log.h

@@ -153,6 +153,8 @@ struct log_t
   static constexpr uint32_t FORMAT_10_8= 0x50687973;
   /** The MariaDB 10.8.0 format with innodb_encrypt_log=ON */
   static constexpr uint32_t FORMAT_ENC_10_8= FORMAT_10_8 | FORMAT_ENCRYPTED;
+  /** The MariaDB 10.11 format with innodb_encrypt_log=ON */
+  static constexpr uint32_t FORMAT_ENC_11= 0xf09f979d;
 
   /** Location of the first checkpoint block */
   static constexpr size_t CHECKPOINT_1= 4096;
@@ -499,12 +501,18 @@ struct log_t
 
   /** Set the log file format. */
   void set_latest_format(bool encrypted) noexcept
-  { format= encrypted ? FORMAT_ENC_10_8 : FORMAT_10_8; }
+  { format= encrypted ? FORMAT_ENC_11 : FORMAT_10_8; }
   /** @return whether the redo log is encrypted */
   bool is_encrypted() const noexcept { return format & FORMAT_ENCRYPTED; }
   /** @return whether the redo log is in the latest format */
   bool is_latest() const noexcept
-  { return (~FORMAT_ENCRYPTED & format) == FORMAT_10_8; }
+  { return format == FORMAT_10_8 || format == FORMAT_ENC_11; }
+  /** @return whether the redo log is in a format that can be recovered */
+  bool is_recoverable() const noexcept
+  {
+    return (format | FORMAT_ENCRYPTED) == FORMAT_ENC_10_8 ||
+      format == FORMAT_ENC_11;
+  }
 
   /** @return capacity in bytes */
   lsn_t capacity() const noexcept { return file_size - START_OFFSET; }

storage/innobase/include/log0recv.h

@@ -252,8 +252,12 @@ struct recv_sys_t
   size_t len;
   /** start offset of non-parsed log records in log_sys.buf */
   size_t offset;
+  /** start offset of the currently parsed mini-transaction */
+  size_t start_offset;
   /** log sequence number of the first non-parsed record */
   lsn_t lsn;
+  /** log sequence number at the start of parse_tail() */
+  lsn_t start_lsn;
   /** log sequence number of the last parsed mini-transaction */
   lsn_t scanned_lsn;
   /** log sequence number at the end of the FILE_CHECKPOINT record, or 0 */
@@ -271,10 +275,17 @@ struct recv_sys_t
   /** iterator to pages, used by parse() */
   map::iterator pages_it;
 
+  /** The allocated size of tmp_buf. The 1+8 extra bytes are
+  needed for FORMAT_ENC_11 in parse(). */
+  static constexpr size_t tmp_buf_size{MTR_SIZE_MAX + 9};
+  /** buffer for decrypting mini-transactions or handling non-contiguous
+  mini-transactions */
+  byte *tmp_buf;
+
   /** Process a record that indicates that a tablespace size is being shrunk.
   @param page_id first page that is not in the file
   @param lsn     log sequence number of the shrink operation */
-  inline void trim(const page_id_t page_id, lsn_t lsn);
+  ATTRIBUTE_COLD void trim(const page_id_t page_id, lsn_t lsn);
 
   /** Undo tablespaces for which truncate has been logged
   (indexed by page_id_t::space() - srv_undo_space_id_start) */
@@ -290,27 +301,53 @@ struct recv_sys_t
   /** The contents of the doublewrite buffer */
   recv_dblwr_t dblwr;
 
-  __attribute__((warn_unused_result)) 
+  /** Free tmp_buf after the log will no longer be parsed. */
+  void tmp_free() noexcept;
+
+  __attribute__((warn_unused_result))
   inline dberr_t read(os_offset_t offset, span<byte> buf);
   inline size_t files_size();
   void close_files();
 
   /** Advance pages_it if it matches the iterator */
-  void pages_it_invalidate(const map::iterator &p)
+  void pages_it_invalidate(const map::iterator &p) noexcept
   {
     mysql_mutex_assert_owner(&mutex);
     if (pages_it == p)
       pages_it++;
   }
   /** Invalidate pages_it if it points to the given tablespace */
-  void pages_it_invalidate(uint32_t space_id)
+  void pages_it_invalidate(uint32_t space_id) noexcept
   {
     mysql_mutex_assert_owner(&mutex);
     if (pages_it != pages.end() && pages_it->first.space() == space_id)
       pages_it= pages.end();
   }
 
 private:
+  /** In parse_tail<storing=NO>(), handle INIT_PAGE or FREE_PAGE
+  @param id        page that is being initialized or freed */
+  void parse_init(const page_id_t id) noexcept;
+
+  /** Handle WRITE to FSP_SPACE_SIZE and FSP_SPACE_FLAGS.
+  @param id       tablespace header page
+  @param b        log record snippet
+  @param size     whether FSP_SPACE_SIZE is being changed
+  @param flags    whether FSP_SPACE_FLAGS is being changed */
+  void parse_page0(const page_id_t id, const byte *b, bool size, bool flags)
+    noexcept;
+
+  /** @return whether parse_store() needs to be invoked
+  @param space_id  tablespace identifier */
+  bool parse_store_if_exists(uint32_t space_id) const noexcept;
+
+  /** Store a parsed log record.
+  @param id          page identifier
+  @param l           log record
+  @param size        size of the log record
+  @return whether we ran out of memory */
+  bool parse_store(const page_id_t id, const byte *l, size_t size) noexcept;
+
   /** Attempt to initialize a page based on redo log records.
   @param p        iterator
   @param mtr      mini-transaction
@@ -381,13 +418,10 @@ struct recv_sys_t
 
   /** Register a redo log snippet for a page.
   @param it       page iterator
-  @param start_lsn start LSN of the mini-transaction
-  @param lsn      @see mtr_t::commit_lsn()
   @param l        redo log snippet
   @param len      length of l, in bytes
   @return whether we ran out of memory */
-  bool add(map::iterator it, lsn_t start_lsn, lsn_t lsn,
-           const byte *l, size_t len);
+  bool add(map::iterator it, const byte *l, size_t len);
 
   /** Parsing result */
   enum parse_mtr_result {
@@ -397,42 +431,67 @@ struct recv_sys_t
     PREMATURE_EOF,
     /** the end of the log was reached */
     GOT_EOF,
-    /** parse<true>(l, false) ran out of memory */
+    /** parse<YES>(l, false) ran out of memory */
     GOT_OOM
   };
 
   /** Whether to store parsed log records */
   enum store{NO,BACKUP,YES};
 
 private:
-  /** Parse and register one log_t::FORMAT_10_8 mini-transaction.
+  /** Parse and register one mini-transaction.
+  @tparam source    type of log data source
   @tparam storing   whether to store the records
+  @tparam format    log record format (log_sys.format)
   @param  l         log data source
   @param  if_exists if store: whether to check if the tablespace exists */
-  template<typename source,store storing>
-  inline parse_mtr_result parse(source &l, bool if_exists) noexcept;
-
-  /** Rewind a mini-transaction when parse() runs out of memory.
-  @param  l         log data source
-  @param  begin     start of the mini-transaction */
-  template<typename source>
-  ATTRIBUTE_COLD void rewind(source &l, source &begin) noexcept;
+  template<typename source,store storing,uint32_t format>
+  inline __attribute__((always_inline))
+  parse_mtr_result parse(source l, bool if_exists) noexcept;
+
+  /** Report that multi-batch recovery is needed.
+  @retval GOT_OOM   always */
+  parse_mtr_result parse_oom() noexcept;
+
+  /** Parse and register one mini-transaction.
+  @tparam ENC_10_8    whether this in log_t::FORMAT_ENC_10_8
+  @tparam storing     whether to store the records
+  @param  begin       start of the mini-transaction
+  @param  if_exists   if store: whether to check if the tablespace exists
+  @param  size        size of the mini-transaction
+  @retval OK          on success
+  @retval GOT_EOF     on corruption
+  @retval GOT_OOM     if we ran out of memory for recv_sys.pages */
+  template<bool ENC_10_8,recv_sys_t::store storing>
+  parse_mtr_result parse_tail(const byte *begin, bool if_exists, size_t size)
+    noexcept;
+
+  /** Rewind a mini-transaction when parse_tail() runs out of memory.
+  @param  begin     start of the mini-transaction
+  @param  end       start of the first unprocessed record */
+  ATTRIBUTE_COLD void rewind(const byte *begin, const byte *end) noexcept;
 
   /** Report progress in terms of LSN or pages remaining */
   ATTRIBUTE_COLD void report_progress() const;
-public:
-  /** Parse and register one log_t::FORMAT_10_8 mini-transaction,
+  /** Parse and register a mini-transaction,
   without handling any log_sys.is_mmap() buffer wrap-around.
   @tparam storing   whether to store the records
+  @tparam format    log_sys.format
   @param  if_exists storing=YES: whether to check if the tablespace exists */
-  template<store storing>
-  static parse_mtr_result parse_mtr(bool if_exists) noexcept;
-  /** Parse and register one log_t::FORMAT_10_8 mini-transaction,
+  template<store storing,uint32_t format>
+  static parse_mtr_result parse_mtr(bool if_exists);
+public:
+  /** Parse and register a mini-transaction,
   handling log_sys.is_mmap() buffer wrap-around.
   @tparam storing   whether to store the records
+  @tparam format    log_sys.format
   @param  if_exists storing=YES: whether to check if the tablespace exists */
-  template<store storing>
-  static parse_mtr_result parse_mmap(bool if_exists) noexcept;
+  template<store storing,uint32_t format>
+  static parse_mtr_result parse_mmap(bool if_exists);
+  /** mini-transaction parser */
+  using parser= parse_mtr_result(*)(bool if_exists);
+  /** @return the parsing function for mariadb-backup --backup */
+  static parser get_backup_parser() noexcept;
 
   /** Erase log records for a page. */
   void erase(map::iterator p);
@@ -462,8 +521,9 @@ struct recv_sys_t
 
   /** Flag data file corruption during recovery. */
   ATTRIBUTE_COLD void set_corrupt_fs() noexcept;
-  /** Flag log file corruption during recovery. */
-  ATTRIBUTE_COLD void set_corrupt_log() noexcept;
+  /** Flag log file corruption during recovery.
+  @retval GOT_EOF   always */
+  ATTRIBUTE_COLD parse_mtr_result set_corrupt_log() noexcept;
 
   /** @return whether data file corruption was found */
   bool is_corrupt_fs() const { return UNIV_UNLIKELY(found_corrupt_fs); }