MDEV-9422: Checksum errors on restart when killing busy instance that uses encrypted XtraDB tables

Fix incorrectly merged files on innodb_plugin.

storage/innobase/log/log0crypt.cc

@@ -36,6 +36,8 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 
 #include "my_crypt.h"
 
+/* Used for debugging */
+// #define DEBUG_CRYPT 1
 #define UNENCRYPTED_KEY_VER	0
 
 /* If true, enable redo log encryption. */
@@ -97,16 +99,24 @@ get_crypt_info(
 {
 	/* so that no one is modifying array while we search */
 	ut_ad(mutex_own(&(log_sys->mutex)));
+	size_t items = crypt_info.size();
 
 	/* a log block only stores 4-bytes of checkpoint no */
 	checkpoint_no &= 0xFFFFFFFF;
-	for (size_t i = 0; i < crypt_info.size(); i++) {
+	for (size_t i = 0; i < items; i++) {
 		struct crypt_info_t* it = &crypt_info[i];
 
 		if (it->checkpoint_no == checkpoint_no) {
 			return it;
 		}
 	}
+
+	/* If checkpoint contains more than one key and we did not
+	find the correct one use the first one. */
+	if (items) {
+		return (&crypt_info[0]);
+	}
+
 	return NULL;
 }
 
@@ -131,7 +141,8 @@ log_blocks_crypt(
 	const byte* block,	/*!< in: blocks before encrypt/decrypt*/
 	ulint size,		/*!< in: size of block */
 	byte* dst_block,	/*!< out: blocks after encrypt/decrypt */
-	int what)		/*!< in: encrypt or decrypt*/
+	int what,		/*!< in: encrypt or decrypt*/
+	const crypt_info_t* crypt_info) /*!< in: crypt info or NULL */
 {
 	byte *log_block = (byte*)block;
 	Crypt_result rc = MY_AES_OK;
@@ -146,7 +157,8 @@ log_blocks_crypt(
 		lsn_t log_block_start_lsn = log_block_get_start_lsn(
 			lsn, log_block_no);
 
-		const crypt_info_t* info = get_crypt_info(log_block);
+		const crypt_info_t* info = crypt_info == NULL ? get_crypt_info(log_block) :
+			crypt_info;
 #ifdef DEBUG_CRYPT
 		fprintf(stderr,
 			"%s %lu chkpt: %lu key: %u lsn: %lu\n",
@@ -301,7 +313,7 @@ log_blocks_encrypt(
 	const ulint size,		/*!< in: size of blocks, must be multiple of a log block */
 	byte* dst_block)		/*!< out: blocks after encryption */
 {
-	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT);
+	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT, NULL);
 }
 
 /*********************************************************************//**
@@ -364,14 +376,16 @@ log_encrypt_before_write(
 		return;
 	}
 
-	if (info->key_version == UNENCRYPTED_KEY_VER) {
+	/* If the key is not encrypted or user has requested not to
+	encrypt, do not change log block. */
+	if (info->key_version == UNENCRYPTED_KEY_VER || !srv_encrypt_log) {
 		return;
 	}
 
 	byte* dst_frame = (byte*)malloc(size);
 
 	//encrypt log blocks content
-	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT);
+	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		ut_ad(block[0] == dst_frame[0]);
@@ -397,7 +411,7 @@ log_decrypt_after_read(
 	byte* dst_frame = (byte*)malloc(size);
 
 	// decrypt log blocks content
-	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT);
+	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
 
 	if (result == MY_AES_OK) {
 		memcpy(frame, dst_frame, size);

storage/innobase/log/log0recv.cc

@@ -2,7 +2,7 @@
 
 Copyright (c) 1997, 2015, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2013, 2015, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2013, 2016, MariaDB Corporation. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -2666,6 +2666,7 @@ recv_scan_log_recs(
 	ibool		finished;
 	ulint		data_len;
 	ibool		more_data;
+	bool		maybe_encrypted=false;
 
 	ut_ad(start_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
 	ut_ad(len % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -2680,6 +2681,8 @@ recv_scan_log_recs(
 	*err = DB_SUCCESS;
 
 	do {
+		log_crypt_err_t log_crypt_err;
+
 		no = log_block_get_hdr_no(log_block);
 		/*
 		fprintf(stderr, "Log block header no %lu\n", no);
@@ -2689,7 +2692,6 @@ recv_scan_log_recs(
 		*/
 		if (no != log_block_convert_lsn_to_no(scanned_lsn)
 		    || !log_block_checksum_is_ok_or_old_format(log_block, true)) {
-			log_crypt_err_t log_crypt_err;
 
 			if (no == log_block_convert_lsn_to_no(scanned_lsn)
 			    && !log_block_checksum_is_ok_or_old_format(
@@ -2707,12 +2709,14 @@ recv_scan_log_recs(
 						log_block));
 			}
 
+			maybe_encrypted = log_crypt_block_maybe_encrypted(log_block,
+					&log_crypt_err);
+
 			/* Garbage or an incompletely written log block */
 
 			finished = TRUE;
 
-			if (log_crypt_block_maybe_encrypted(log_block,
-					&log_crypt_err)) {
+			if (maybe_encrypted) {
 				/* Log block maybe encrypted finish processing*/
 				log_crypt_print_error(log_crypt_err);
 				*err = DB_ERROR;