MDEV-8815: InnoDB should refuse to start if crash recovery fails instead of asserting

Added error handling to crash recovery so that we stop instead of
asserting.

4 files changed, 80 insertions, 38 deletions

diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 8ede49d4ecc..e4a06fcc532 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -275,7 +275,7 @@ recv_sys_var_init(void);
 Empties the hash table of stored log records, applying them to appropriate
 pages. */
 UNIV_INTERN
-void
+dberr_t
 recv_apply_hashed_log_recs(
 /*=======================*/
 	ibool	allow_ibuf);	/*!< in: if TRUE, also ibuf operations are
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 286fcbb8ae0..b7a7b09bcb1 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -442,9 +442,10 @@ recv_sys_init(
 }
 
 /********************************************************//**
-Empties the hash table when it has been fully processed. */
+Empties the hash table when it has been fully processed.
+@return DB_SUCCESS when successfull or DB_ERROR when fails. */
 static
-void
+dberr_t
 recv_sys_empty_hash(void)
 /*=====================*/
 {
@@ -458,13 +459,15 @@ recv_sys_empty_hash(void)
 			" log records on it %lu\n",
 			(ulong) recv_sys->n_addrs,
 			(ulong) recv_max_parsed_page_no);
-		ut_error;
+		return DB_ERROR;
 	}
 
 	hash_table_free(recv_sys->addr_hash);
 	mem_heap_empty(recv_sys->heap);
 
 	recv_sys->addr_hash = hash_create(buf_pool_get_curr_size() / 512);
+
+	return DB_SUCCESS;
 }
 
 #ifndef UNIV_HOTBACKUP
@@ -1779,9 +1782,10 @@ recv_read_in_area(
 
 /*******************************************************************//**
 Empties the hash table of stored log records, applying them to appropriate
-pages. */
+pages.
+@return DB_SUCCESS when successfull or DB_ERROR when fails. */
 UNIV_INTERN
-void
+dberr_t
 recv_apply_hashed_log_recs(
 /*=======================*/
 	ibool	allow_ibuf)	/*!< in: if TRUE, also ibuf operations are
@@ -1798,6 +1802,7 @@ recv_apply_hashed_log_recs(
 	ulint	i;
 	ibool	has_printed	= FALSE;
 	mtr_t	mtr;
+	dberr_t err = DB_SUCCESS;
 loop:
 	mutex_enter(&(recv_sys->mutex));
 
@@ -1931,13 +1936,15 @@ loop:
 	recv_sys->apply_log_recs = FALSE;
 	recv_sys->apply_batch_on = FALSE;
 
-	recv_sys_empty_hash();
+	err = recv_sys_empty_hash();
 
 	if (has_printed) {
 		fprintf(stderr, "InnoDB: Apply batch completed\n");
 	}
 
 	mutex_exit(&(recv_sys->mutex));
+
+	return err;
 }
 #else /* !UNIV_HOTBACKUP */
 /*******************************************************************//**
@@ -2258,7 +2265,6 @@ recv_report_corrupt_log(
 	if (!srv_force_recovery) {
 		fputs("InnoDB: Set innodb_force_recovery"
 		      " to ignore this error.\n", stderr);
-		ut_error;
 	}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -2281,9 +2287,11 @@ static
 ibool
 recv_parse_log_recs(
 /*================*/
-	ibool	store_to_hash)	/*!< in: TRUE if the records should be stored
+	ibool	store_to_hash,	/*!< in: TRUE if the records should be stored
 				to the hash table; this is set to FALSE if just
 				debug checking is needed */
+	dberr_t* err)		/*!< out: DB_SUCCESS if successfull,
+				DB_ERROR if parsing fails. */
 {
 	byte*	ptr;
 	byte*	end_ptr;
@@ -2392,7 +2400,8 @@ loop:
 						(ulint) type, space,
 						(char*)(body + 2));
 
-					ut_error;
+					*err = DB_ERROR;
+					return(FALSE);
 				}
 			}
 #endif
@@ -2688,17 +2697,18 @@ recv_scan_log_recs(
 
 			if (log_crypt_block_maybe_encrypted(log_block,
 					&log_crypt_err)) {
-				/* Log block maybe encrypted */
+				/* Log block maybe encrypted finish processing*/
 				log_crypt_print_error(log_crypt_err);
 				*err = DB_ERROR;
 				return (TRUE);
 			}
 
-			/* Crash if we encounter a garbage log block */
+			/* Stop if we encounter a garbage log block */
 			if (!srv_force_recovery) {
 				fputs("InnoDB: Set innodb_force_recovery"
 				      " to ignore this error.\n", stderr);
-				ut_error;
+				*err = DB_ERROR;
+				return (TRUE);
 			}
 
 			break;
@@ -2736,7 +2746,8 @@ recv_scan_log_recs(
 			/* This is not really an error, but currently
 			we stop here in the debug version: */
 
-			ut_error;
+			*err = DB_ERROR;
+			return (TRUE);
 #endif
 			break;
 		}
@@ -2802,7 +2813,8 @@ recv_scan_log_recs(
 					      " innodb_force_recovery"
 					      " to ignore this error.\n",
 					      stderr);
-					ut_error;
+					*err = DB_ERROR;
+					return (TRUE);
 				}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -2844,7 +2856,11 @@ recv_scan_log_recs(
 	if (more_data && !recv_sys->found_corrupt_log) {
 		/* Try to parse more log records */
 
-		recv_parse_log_recs(store_to_hash);
+		recv_parse_log_recs(store_to_hash, err);
+
+		if (*err != DB_SUCCESS) {
+			return (TRUE);
+		}
 
 #ifndef UNIV_HOTBACKUP
 		if (store_to_hash
@@ -2856,7 +2872,12 @@ recv_scan_log_recs(
 			log yet: they would be produced by ibuf
 			operations */
 
-			recv_apply_hashed_log_recs(FALSE);
+			*err = recv_apply_hashed_log_recs(FALSE);
+
+			if (*err != DB_SUCCESS) {
+				/* Finish processing because of error */
+				return (TRUE);
+			}
 		}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -3174,7 +3195,7 @@ recv_recovery_from_checkpoint_start_func(
 #ifdef UNIV_LOG_ARCHIVE
 		lsn_t	old_scanned_lsn	= recv_sys->scanned_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
-		dberr_t err;
+		dberr_t err = DB_SUCCESS;
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
 			&group_scanned_lsn, &err);
@@ -3276,7 +3297,7 @@ recv_recovery_from_checkpoint_start_func(
 
 		/* No harm in trying to do RO access. */
 		if (!srv_read_only_mode) {
-			ut_error;
+			return (DB_READ_ONLY);
 		}
 
 		return(DB_ERROR);
diff --git a/storage/xtradb/include/log0recv.h b/storage/xtradb/include/log0recv.h
index 805b6c66768..b23a140aac2 100644
--- a/storage/xtradb/include/log0recv.h
+++ b/storage/xtradb/include/log0recv.h
@@ -312,7 +312,7 @@ recv_sys_var_init(void);
 Empties the hash table of stored log records, applying them to appropriate
 pages. */
 UNIV_INTERN
-void
+dberr_t
 recv_apply_hashed_log_recs(
 /*=======================*/
 	ibool	allow_ibuf);	/*!< in: if TRUE, also ibuf operations are
diff --git a/storage/xtradb/log/log0recv.cc b/storage/xtradb/log/log0recv.cc
index af3b7db735b..829c0c8d75d 100644
--- a/storage/xtradb/log/log0recv.cc
+++ b/storage/xtradb/log/log0recv.cc
@@ -446,9 +446,10 @@ recv_sys_init(
 }
 
 /********************************************************//**
-Empties the hash table when it has been fully processed. */
+Empties the hash table when it has been fully processed.
+@return DB_SUCCESS when successfull or DB_ERROR when fails. */
 static
-void
+dberr_t
 recv_sys_empty_hash(void)
 /*=====================*/
 {
@@ -462,13 +463,15 @@ recv_sys_empty_hash(void)
 			" log records on it %lu\n",
 			(ulong) recv_sys->n_addrs,
 			(ulong) recv_max_parsed_page_no);
-		ut_error;
+		return DB_ERROR;
 	}
 
 	hash_table_free(recv_sys->addr_hash);
 	mem_heap_empty(recv_sys->heap);
 
 	recv_sys->addr_hash = hash_create(buf_pool_get_curr_size() / 512);
+
+	return DB_SUCCESS;
 }
 
 #ifndef UNIV_HOTBACKUP
@@ -1847,9 +1850,10 @@ recv_read_in_area(
 
 /*******************************************************************//**
 Empties the hash table of stored log records, applying them to appropriate
-pages. */
+pages.
+@return DB_SUCCESS when successfull or DB_ERROR when fails. */
 UNIV_INTERN
-void
+dberr_t
 recv_apply_hashed_log_recs(
 /*=======================*/
 	ibool	allow_ibuf)	/*!< in: if TRUE, also ibuf operations are
@@ -1866,6 +1870,7 @@ recv_apply_hashed_log_recs(
 	ulint	i;
 	ibool	has_printed	= FALSE;
 	mtr_t	mtr;
+	dberr_t err = DB_SUCCESS;
 loop:
 	mutex_enter(&(recv_sys->mutex));
 
@@ -1999,13 +2004,15 @@ loop:
 	recv_sys->apply_log_recs = FALSE;
 	recv_sys->apply_batch_on = FALSE;
 
-	recv_sys_empty_hash();
+	err = recv_sys_empty_hash();
 
 	if (has_printed) {
 		fprintf(stderr, "InnoDB: Apply batch completed\n");
 	}
 
 	mutex_exit(&(recv_sys->mutex));
+
+	return err;
 }
 #else /* !UNIV_HOTBACKUP */
 /*******************************************************************//**
@@ -2328,7 +2335,6 @@ recv_report_corrupt_log(
 	if (!srv_force_recovery) {
 		fputs("InnoDB: Set innodb_force_recovery"
 		      " to ignore this error.\n", stderr);
-		ut_error;
 	}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -2351,9 +2357,11 @@ static
 ibool
 recv_parse_log_recs(
 /*================*/
-	ibool	store_to_hash)	/*!< in: TRUE if the records should be stored
+	ibool	store_to_hash,	/*!< in: TRUE if the records should be stored
 				to the hash table; this is set to FALSE if just
 				debug checking is needed */
+	dberr_t* err)		/*!< out: DB_SUCCESS if successfull,
+				DB_ERROR if parsing fails. */
 {
 	byte*	ptr;
 	byte*	end_ptr;
@@ -2462,7 +2470,8 @@ loop:
 						(ulint) type, space,
 						(char*)(body + 2));
 
-					ut_error;
+					*err = DB_ERROR;
+					return(FALSE);
 				}
 			}
 #endif
@@ -2758,17 +2767,18 @@ recv_scan_log_recs(
 
 			if (log_crypt_block_maybe_encrypted(log_block,
 					&log_crypt_err)) {
-				/* Log block maybe encrypted */
+				/* Log block maybe encrypted finish processing*/
 				log_crypt_print_error(log_crypt_err);
 				*err = DB_ERROR;
 				return (TRUE);
 			}
 
-			/* Crash if we encounter a garbage log block */
+			/* Stop if we encounter a garbage log block */
 			if (!srv_force_recovery) {
 				fputs("InnoDB: Set innodb_force_recovery"
 				      " to ignore this error.\n", stderr);
-				ut_error;
+				*err = DB_ERROR;
+				return (TRUE);
 			}
 
 			break;
@@ -2806,7 +2816,8 @@ recv_scan_log_recs(
 			/* This is not really an error, but currently
 			we stop here in the debug version: */
 
-			ut_error;
+			*err = DB_ERROR;
+			return (TRUE);
 #endif
 			break;
 		}
@@ -2872,7 +2883,8 @@ recv_scan_log_recs(
 					      " innodb_force_recovery"
 					      " to ignore this error.\n",
 					      stderr);
-					ut_error;
+					*err = DB_ERROR;
+					return (TRUE);
 				}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -2914,7 +2926,11 @@ recv_scan_log_recs(
 	if (more_data && !recv_sys->found_corrupt_log) {
 		/* Try to parse more log records */
 
-		recv_parse_log_recs(store_to_hash);
+		recv_parse_log_recs(store_to_hash, err);
+
+		if (*err != DB_SUCCESS) {
+			return (TRUE);
+		}
 
 #ifndef UNIV_HOTBACKUP
 		if (store_to_hash
@@ -2926,7 +2942,12 @@ recv_scan_log_recs(
 			log yet: they would be produced by ibuf
 			operations */
 
-			recv_apply_hashed_log_recs(FALSE);
+			*err = recv_apply_hashed_log_recs(FALSE);
+
+			if (*err != DB_SUCCESS) {
+				/* Finish processing because of error */
+				return (TRUE);
+			}
 		}
 #endif /* !UNIV_HOTBACKUP */
 
@@ -3267,7 +3288,7 @@ recv_recovery_from_checkpoint_start_func(
 #ifdef UNIV_LOG_ARCHIVE
 		lsn_t	old_scanned_lsn	= recv_sys->scanned_lsn;
 #endif /* UNIV_LOG_ARCHIVE */
-		dberr_t err;
+		dberr_t err = DB_SUCCESS;
 
 		recv_group_scan_log_recs(group, &contiguous_lsn,
 			&group_scanned_lsn, &err);
@@ -3369,7 +3390,7 @@ recv_recovery_from_checkpoint_start_func(
 
 		/* No harm in trying to do RO access. */
 		if (!srv_read_only_mode) {
-			ut_error;
+			return (DB_READ_ONLY);
 		}
 
 		return(DB_ERROR);