MDEV-12353: Make UNDO_APPEND more robust

This is a follow-up to commit 84e3f9ce
that introduced the EXTENDED log record of UNDO_APPEND subtype.

mtr_t::undo_append(): Accurately enforce the mtr_buf_t::MAX_DATA_SIZE
limit. Also, replace mtr_buf_t::push() with simpler code, to append 1 byte
to the log.

log_phys_t::undo_append(): Return whether the page was found to
be in an inconsistent state.

log_phys_t::apply(): If corruption was noticed, stop applying log
unless innodb_force_recovery is set.

storage/innobase/include/mtr0log.h

@@ -523,7 +523,7 @@ inline void mtr_t::free(const page_id_t id)
 
 /** Write an EXTENDED log record.
 @param block  buffer pool page
-@param type	extended record subtype; @see mrec_ext_t */
+@param type   extended record subtype; @see mrec_ext_t */
 inline void mtr_t::log_write_extended(const buf_block_t &block, byte type)
 {
   set_modified();
@@ -615,7 +615,7 @@ inline void mtr_t::undo_append(const buf_block_t &block,
   set_modified();
   if (m_log_mode != MTR_LOG_ALL)
     return;
-  const bool small= len < mtr_buf_t::MAX_DATA_SIZE - (3 + 3 + 5 + 5);
+  const bool small= len + 1 < mtr_buf_t::MAX_DATA_SIZE - (1 + 3 + 3 + 5 + 5);
   byte *end= log_write<EXTENDED>(block.page.id, &block.page, len + 1, small);
   if (UNIV_LIKELY(small))
   {
@@ -626,8 +626,9 @@ inline void mtr_t::undo_append(const buf_block_t &block,
   else
   {
     m_log.close(end);
-    byte type= UNDO_APPEND;
-    m_log.push(&type, 1);
+    end= m_log.open(1);
+    *end++= UNDO_APPEND;
+    m_log.close(end);
     m_log.push(static_cast<const byte*>(data), static_cast<uint32_t>(len));
   }
   m_last_offset= FIL_PAGE_TYPE;

storage/innobase/include/mtr0mtr.h

@@ -551,7 +551,7 @@ struct mtr_t {
 
   /** Write an EXTENDED log record.
   @param block  buffer pool page
-  @param type	extended record subtype; @see mrec_ext_t */
+  @param type   extended record subtype; @see mrec_ext_t */
   inline void log_write_extended(const buf_block_t &block, byte type);
 
   /** Prepare to write the mini-transaction log to the redo log buffer.

storage/innobase/log/log0recv.cc

@@ -157,8 +157,9 @@ struct log_phys_t : public log_rec_t
   @see mtr_t::undo_append()
   @param block   undo log page
   @param data    undo log record
-  @param len     length of the undo log record */
-  static void undo_append(const buf_block_t &block, const byte *data,
+  @param len     length of the undo log record
+  @return whether the operation failed (inconcistency was noticed) */
+  static bool undo_append(const buf_block_t &block, const byte *data,
                           size_t len)
   {
     ut_ad(len > 2);
@@ -170,7 +171,7 @@ struct log_phys_t : public log_rec_t
     {
       ib::error() << "Not applying UNDO_APPEND due to corruption on "
                   << block.page.id;
-      return;
+      return true;
     }
 
     byte *p= block.frame + free;
@@ -180,6 +181,7 @@ struct log_phys_t : public log_rec_t
     memcpy(p, data, len);
     p+= len;
     mach_write_to_2(p, free);
+    return false;
   }
 
   /** The status of apply() */
@@ -307,12 +309,17 @@ struct log_phys_t : public log_rec_t
         case UNDO_APPEND:
           if (UNIV_UNLIKELY(rlen <= 3))
             goto record_corrupted;
-          undo_append(block, ++l, --rlen);
+          if (undo_append(block, ++l, --rlen) && !srv_force_recovery)
+          {
+            ib::error() << "Set innodb_force_recovery=1 to ignore corruption.";
+            recv_sys.found_corrupt_log= true;
+            return applied;
+          }
           break;
         case DELETE_ROW_FORMAT_REDUNDANT:
           if (UNIV_UNLIKELY(rlen < 2 || rlen > 4))
             goto record_corrupted;
-	  rlen--;
+          rlen--;
           ll= mlog_decode_varint_length(*++l);
           if (UNIV_UNLIKELY(ll != rlen))
             goto record_corrupted;
@@ -321,7 +328,7 @@ struct log_phys_t : public log_rec_t
         case DELETE_ROW_FORMAT_DYNAMIC:
           if (UNIV_UNLIKELY(rlen < 2))
             goto record_corrupted;
-	  rlen--;
+          rlen--;
           ll= mlog_decode_varint_length(*++l);
           if (UNIV_UNLIKELY(ll > 3 || ll >= rlen))
             goto record_corrupted;
@@ -2274,6 +2281,10 @@ static void recv_recover_page(buf_block_t* block, mtr_t& mtr,
 		}
 
 set_start_lsn:
+		if (recv_sys.found_corrupt_log && !srv_force_recovery) {
+			break;
+		}
+
 		if (!start_lsn) {
 			start_lsn = l->start_lsn;
 		}