diff --git a/include/heap.h b/include/heap.h
index d27d6d7ab4073..bf3e8122ad3de 100644
--- a/include/heap.h
+++ b/include/heap.h
@@ -143,6 +143,17 @@ typedef struct st_hp_blob_desc
   uint packlength;   /* 1, 2, 3, or 4: length prefix size */
 } HP_BLOB_DESC;
 
+/*
+  Bits for HP_SHARE::state_changed, modeled on the state.changed bitmaps
+  of Maria and MyISAM (see storage/maria/maria_def.h).  A table marked
+  crashed refuses every read and write with HA_ERR_CRASHED until it is
+  re-created or emptied (hp_clear()).
+*/
+#define HEAP_STATE_CRASHED 1U
+
+#define heap_mark_crashed(share) ((share)->state_changed|= HEAP_STATE_CRASHED)
+#define heap_is_crashed(share)   ((share)->state_changed & HEAP_STATE_CRASHED)
+
 typedef struct st_heap_share
 {
   HP_BLOCK block;
@@ -160,6 +171,7 @@ typedef struct st_heap_share
   uint reclength;			/* Length of one record */
   uint visible;     /* Offset to the flags byte (active/deleted/continuation) */
   uint changed;
+  uint state_changed;                   /* Bitmap of HEAP_STATE_* flags */
   uint keys,max_key_length;
   uint currently_disabled_keys;    /* saved value from "keys" when disabled */
   uint open_count;
diff --git a/include/my_tree.h b/include/my_tree.h
index 9bed28efcb321..2507fb180df83 100644
--- a/include/my_tree.h
+++ b/include/my_tree.h
@@ -48,6 +48,8 @@ typedef uint32 element_count;
 typedef int (*tree_walk_action)(void *,element_count,void *);
 
 typedef enum { free_init, free_free, free_end } TREE_FREE;
+typedef enum { TREE_ERROR_NONE, TREE_ERROR_OOM, TREE_ERROR_DUP_KEY }
+TREE_ERROR;
 typedef int (*tree_element_free)(void*, TREE_FREE, void *);
 
 typedef struct st_tree_element {
@@ -70,6 +72,7 @@ typedef struct st_tree {
   tree_element_free free;
   myf my_flags;
   uint flag;
+  TREE_ERROR error;
 } TREE;
 
 	/* Functions on whole tree */
diff --git a/mysql-test/suite/heap/blob_update_key_rollback.result b/mysql-test/suite/heap/blob_update_key_rollback.result
new file mode 100644
index 0000000000000..e47ede0edb1e0
--- /dev/null
+++ b/mysql-test/suite/heap/blob_update_key_rollback.result
@@ -0,0 +1,186 @@
+#
+# Index consistency after a failed UPDATE on a HEAP table with BLOB
+#
+# heap_update() moves changed key entries to the new key values
+# before writing the new blob chains.  When a blob chain write then
+# fails with "table is full", the key entries must be moved back to
+# the old values, otherwise the index refers to records whose
+# columns no longer match the indexed values.
+#
+SET @save_max_heap= @@max_heap_table_size;
+SET max_heap_table_size = 16384;
+#
+# Hash index: single-row UPDATE fails writing the new blob chain
+#
+CREATE TABLE t1 (
+id INT,
+k VARCHAR(20),
+b BLOB,
+KEY k1 (k)
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_10;
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 1;
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# The row must still be found through the index under the old value
+SELECT id, k, LENGTH(b) FROM t1 FORCE INDEX(k1) WHERE k = 'key1';
+id	k	LENGTH(b)
+1	key1	300
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+COUNT(*)
+0
+DROP TABLE t1;
+#
+# BTREE index: same failure, rb-tree entries must be restored
+#
+CREATE TABLE t1 (
+id INT,
+k VARCHAR(20),
+b BLOB,
+KEY k1 (k) USING BTREE
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_10;
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 1;
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SELECT id, k, LENGTH(b) FROM t1 FORCE INDEX(k1) WHERE k = 'key1';
+id	k	LENGTH(b)
+1	key1	300
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+COUNT(*)
+0
+DROP TABLE t1;
+#
+# Two indexed columns changed by the same failing UPDATE:
+# entries in both indexes must be rolled back
+#
+CREATE TABLE t1 (
+id INT,
+k VARCHAR(20),
+m VARCHAR(20),
+b BLOB,
+KEY k1 (k),
+KEY k2 (m) USING BTREE
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), CONCAT('mem', seq),
+REPEAT('a', 300)
+FROM seq_1_to_10;
+UPDATE t1 SET k= 'moved', m= 'gone', b= REPEAT('z', 60000) WHERE id = 2;
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SELECT id, k, m FROM t1 FORCE INDEX(k1) WHERE k = 'key2';
+id	k	m
+2	key2	mem2
+SELECT id, k, m FROM t1 FORCE INDEX(k2) WHERE m = 'mem2';
+id	k	m
+2	key2	mem2
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+COUNT(*)
+0
+SELECT COUNT(*) FROM t1 FORCE INDEX(k2) WHERE m = 'gone';
+COUNT(*)
+0
+DROP TABLE t1;
+#
+# A second index on a column the UPDATE does NOT touch must be left
+# alone by the rollback: only the changed key is moved back, the
+# unchanged key's entries stay put and keep finding the row.
+#
+CREATE TABLE t1 (
+id INT,
+k VARCHAR(20),
+fixed VARCHAR(20),
+b BLOB,
+KEY k1 (k),
+KEY k2 (fixed)
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), CONCAT('fix', seq),
+REPEAT('a', 300)
+FROM seq_1_to_10;
+# Only k and b change; 'fixed' is untouched
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 3;
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Changed key rolled back to old value
+SELECT id, k, fixed FROM t1 FORCE INDEX(k1) WHERE k = 'key3';
+id	k	fixed
+3	key3	fix3
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+COUNT(*)
+0
+# Untouched key still resolves the row through its own index
+SELECT id, k, fixed FROM t1 FORCE INDEX(k2) WHERE fixed = 'fix3';
+id	k	fixed
+3	key3	fix3
+DROP TABLE t1;
+#
+# Multi-row UPDATE that fails part-way: rows updated before the
+# failure keep their new key values, the failing row keeps its old
+# ones, and every row must be reachable through the index by its
+# current key value
+#
+CREATE TABLE t1 (
+id INT,
+k VARCHAR(20),
+b BLOB,
+KEY k1 (k)
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_15;
+# Each row grows its blob; the table fills up after some rows
+UPDATE t1 SET k= CONCAT('new', id), b= REPEAT('z', 1500);
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Every row must be found via the index under its current key value
+SELECT t.id FROM t1 t
+WHERE NOT EXISTS (SELECT 1 FROM t1 i FORCE INDEX(k1)
+WHERE i.k = t.k AND i.id = t.id)
+ORDER BY t.id;
+id
+# Old and new key values must partition the table, with the
+# failure hitting neither the first row nor after the last one.
+# The exact fill point depends on platform record layout, so only
+# the invariants are checked.
+SELECT COUNT(*) AS total,
+COUNT(IF(k LIKE 'new%', 1, NULL)) > 0 AS some_updated,
+COUNT(IF(k LIKE 'key%', 1, NULL)) > 0 AS some_old,
+COUNT(IF(k LIKE 'new%', 1, NULL)) +
+COUNT(IF(k LIKE 'key%', 1, NULL)) = COUNT(*) AS partitioned
+FROM t1;
+total	some_updated	some_old	partitioned
+15	1	1	1
+DROP TABLE t1;
+#
+# Table with no indexes at all: the failed blob write must leave the
+# record intact with nothing to roll back in the (empty) key set
+#
+CREATE TABLE t1 (
+id INT,
+b BLOB
+) ENGINE=HEAP;
+INSERT INTO t1 SELECT seq, REPEAT('a', 300) FROM seq_1_to_10;
+UPDATE t1 SET b= REPEAT('z', 60000) WHERE id = 1;
+ERROR HY000: The table 't1' is full
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SELECT id, LENGTH(b) FROM t1 WHERE id = 1;
+id	LENGTH(b)
+1	300
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+10
+DROP TABLE t1;
+SET max_heap_table_size = @save_max_heap;
diff --git a/mysql-test/suite/heap/blob_update_key_rollback.test b/mysql-test/suite/heap/blob_update_key_rollback.test
new file mode 100644
index 0000000000000..b2ba7136cdb9a
--- /dev/null
+++ b/mysql-test/suite/heap/blob_update_key_rollback.test
@@ -0,0 +1,179 @@
+--source include/have_sequence.inc
+
+--echo #
+--echo # Index consistency after a failed UPDATE on a HEAP table with BLOB
+--echo #
+--echo # heap_update() moves changed key entries to the new key values
+--echo # before writing the new blob chains.  When a blob chain write then
+--echo # fails with "table is full", the key entries must be moved back to
+--echo # the old values, otherwise the index refers to records whose
+--echo # columns no longer match the indexed values.
+--echo #
+
+SET @save_max_heap= @@max_heap_table_size;
+SET max_heap_table_size = 16384;
+
+--echo #
+--echo # Hash index: single-row UPDATE fails writing the new blob chain
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  k VARCHAR(20),
+  b BLOB,
+  KEY k1 (k)
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_10;
+
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 1;
+
+CHECK TABLE t1;
+--echo # The row must still be found through the index under the old value
+SELECT id, k, LENGTH(b) FROM t1 FORCE INDEX(k1) WHERE k = 'key1';
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+DROP TABLE t1;
+
+--echo #
+--echo # BTREE index: same failure, rb-tree entries must be restored
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  k VARCHAR(20),
+  b BLOB,
+  KEY k1 (k) USING BTREE
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_10;
+
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 1;
+
+CHECK TABLE t1;
+SELECT id, k, LENGTH(b) FROM t1 FORCE INDEX(k1) WHERE k = 'key1';
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+DROP TABLE t1;
+
+--echo #
+--echo # Two indexed columns changed by the same failing UPDATE:
+--echo # entries in both indexes must be rolled back
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  k VARCHAR(20),
+  m VARCHAR(20),
+  b BLOB,
+  KEY k1 (k),
+  KEY k2 (m) USING BTREE
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), CONCAT('mem', seq),
+                      REPEAT('a', 300)
+FROM seq_1_to_10;
+
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET k= 'moved', m= 'gone', b= REPEAT('z', 60000) WHERE id = 2;
+
+CHECK TABLE t1;
+SELECT id, k, m FROM t1 FORCE INDEX(k1) WHERE k = 'key2';
+SELECT id, k, m FROM t1 FORCE INDEX(k2) WHERE m = 'mem2';
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+SELECT COUNT(*) FROM t1 FORCE INDEX(k2) WHERE m = 'gone';
+DROP TABLE t1;
+
+--echo #
+--echo # A second index on a column the UPDATE does NOT touch must be left
+--echo # alone by the rollback: only the changed key is moved back, the
+--echo # unchanged key's entries stay put and keep finding the row.
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  k VARCHAR(20),
+  fixed VARCHAR(20),
+  b BLOB,
+  KEY k1 (k),
+  KEY k2 (fixed)
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), CONCAT('fix', seq),
+                      REPEAT('a', 300)
+FROM seq_1_to_10;
+
+--echo # Only k and b change; 'fixed' is untouched
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET k= 'moved', b= REPEAT('z', 60000) WHERE id = 3;
+
+CHECK TABLE t1;
+--echo # Changed key rolled back to old value
+SELECT id, k, fixed FROM t1 FORCE INDEX(k1) WHERE k = 'key3';
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE k = 'moved';
+--echo # Untouched key still resolves the row through its own index
+SELECT id, k, fixed FROM t1 FORCE INDEX(k2) WHERE fixed = 'fix3';
+DROP TABLE t1;
+
+--echo #
+--echo # Multi-row UPDATE that fails part-way: rows updated before the
+--echo # failure keep their new key values, the failing row keeps its old
+--echo # ones, and every row must be reachable through the index by its
+--echo # current key value
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  k VARCHAR(20),
+  b BLOB,
+  KEY k1 (k)
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, CONCAT('key', seq), REPEAT('a', 300)
+FROM seq_1_to_15;
+
+--echo # Each row grows its blob; the table fills up after some rows
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET k= CONCAT('new', id), b= REPEAT('z', 1500);
+
+CHECK TABLE t1;
+--echo # Every row must be found via the index under its current key value
+SELECT t.id FROM t1 t
+WHERE NOT EXISTS (SELECT 1 FROM t1 i FORCE INDEX(k1)
+                  WHERE i.k = t.k AND i.id = t.id)
+ORDER BY t.id;
+--echo # Old and new key values must partition the table, with the
+--echo # failure hitting neither the first row nor after the last one.
+--echo # The exact fill point depends on platform record layout, so only
+--echo # the invariants are checked.
+SELECT COUNT(*) AS total,
+       COUNT(IF(k LIKE 'new%', 1, NULL)) > 0 AS some_updated,
+       COUNT(IF(k LIKE 'key%', 1, NULL)) > 0 AS some_old,
+       COUNT(IF(k LIKE 'new%', 1, NULL)) +
+       COUNT(IF(k LIKE 'key%', 1, NULL)) = COUNT(*) AS partitioned
+FROM t1;
+DROP TABLE t1;
+
+--echo #
+--echo # Table with no indexes at all: the failed blob write must leave the
+--echo # record intact with nothing to roll back in the (empty) key set
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  b BLOB
+) ENGINE=HEAP;
+
+INSERT INTO t1 SELECT seq, REPEAT('a', 300) FROM seq_1_to_10;
+
+--error ER_RECORD_FILE_FULL
+UPDATE t1 SET b= REPEAT('z', 60000) WHERE id = 1;
+
+CHECK TABLE t1;
+SELECT id, LENGTH(b) FROM t1 WHERE id = 1;
+SELECT COUNT(*) FROM t1;
+DROP TABLE t1;
+
+SET max_heap_table_size = @save_max_heap;
diff --git a/mysql-test/suite/heap/update_key_rollback.result b/mysql-test/suite/heap/update_key_rollback.result
new file mode 100644
index 0000000000000..4c3df32a6f523
--- /dev/null
+++ b/mysql-test/suite/heap/update_key_rollback.result
@@ -0,0 +1,105 @@
+#
+# Index consistency after a failed UPDATE on a HEAP table
+#
+# heap_update() moves the changed key entries to their new key values
+# before it updates the record.  A failure raised after that has to
+# move them back.  An rb-tree node allocation failure used to be
+# reported as a duplicate key, and when the recovery itself cannot
+# restore a key the table is now marked crashed, so that no statement
+# can use an index that no longer describes the data.
+#
+# The keyword lists armed below always carry the inert
+# "heap_test_guard" keyword: the one-shot injection removes its own
+# keyword when it fires, and a keyword list that becomes empty while
+# debugging is on matches every keyword.
+#
+CALL mtr.add_suppression("Index for table .t1. is corrupt");
+#
+# The hash key k1 is moved to the new value, then the rb-tree node
+# allocation for k2 fails once: the failure is reported as out of
+# memory, not as a duplicate, and both keys are moved back
+#
+CREATE TABLE t1 (
+id INT,
+a INT,
+b INT,
+KEY k1 (a),
+KEY k2 (b) USING BTREE
+) ENGINE=HEAP;
+INSERT INTO t1 VALUES (1, 10, 100), (2, 20, 200);
+SET SESSION debug_dbug="+d,once_simulate_tree_insert_oom,heap_test_guard";
+UPDATE t1 SET a= 11, b= 101 WHERE id = 1;
+ERROR HY000: Out of memory.
+SET SESSION debug_dbug="";
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Both indexes must still find the row under its unchanged values
+SELECT id, a, b FROM t1 FORCE INDEX(k1) WHERE a = 10;
+id	a	b
+1	10	100
+SELECT id, a, b FROM t1 FORCE INDEX(k2) WHERE b = 100;
+id	a	b
+1	10	100
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE a = 11;
+COUNT(*)
+0
+SELECT COUNT(*) FROM t1 FORCE INDEX(k2) WHERE b = 101;
+COUNT(*)
+0
+SELECT id, a, b FROM t1 ORDER BY id;
+id	a	b
+1	10	100
+2	20	200
+#
+# A duplicate key is still reported as a duplicate
+#
+CREATE TABLE t2 (
+id INT,
+a INT,
+UNIQUE KEY k1 (a) USING BTREE
+) ENGINE=HEAP;
+INSERT INTO t2 VALUES (1, 10), (2, 20);
+UPDATE t2 SET a= 20 WHERE id = 1;
+ERROR 23000: Duplicate entry '20' for key 'k1'
+CHECK TABLE t2;
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK
+SELECT id, a FROM t2 ORDER BY id;
+id	a
+1	10
+2	20
+DROP TABLE t2;
+#
+# The allocation failure persists, so restoring the old k2 entry
+# fails as well: the already moved k1 no longer describes the
+# record, the table is marked crashed, and every read and write on
+# it is refused until it is emptied
+#
+SET SESSION debug_dbug="+d,simulate_tree_insert_oom,heap_test_guard";
+UPDATE t1 SET a= 11, b= 101 WHERE id = 1;
+ERROR HY000: Out of memory.
+SET SESSION debug_dbug="";
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	error	Corrupt
+SELECT id, a, b FROM t1 ORDER BY id;
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
+SELECT id, a, b FROM t1 FORCE INDEX(k1) WHERE a = 10;
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
+INSERT INTO t1 VALUES (3, 30, 300);
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
+UPDATE t1 SET a= 12 WHERE id = 2;
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
+DELETE FROM t1 WHERE id = 2;
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
+# Emptying the table rebuilds the indexes from nothing
+TRUNCATE TABLE t1;
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+INSERT INTO t1 VALUES (5, 50, 500);
+SELECT id, a, b FROM t1 FORCE INDEX(k2) WHERE b = 500;
+id	a	b
+5	50	500
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/update_key_rollback.test b/mysql-test/suite/heap/update_key_rollback.test
new file mode 100644
index 0000000000000..374ac6f703364
--- /dev/null
+++ b/mysql-test/suite/heap/update_key_rollback.test
@@ -0,0 +1,99 @@
+--source include/have_debug.inc
+
+--echo #
+--echo # Index consistency after a failed UPDATE on a HEAP table
+--echo #
+--echo # heap_update() moves the changed key entries to their new key values
+--echo # before it updates the record.  A failure raised after that has to
+--echo # move them back.  An rb-tree node allocation failure used to be
+--echo # reported as a duplicate key, and when the recovery itself cannot
+--echo # restore a key the table is now marked crashed, so that no statement
+--echo # can use an index that no longer describes the data.
+--echo #
+--echo # The keyword lists armed below always carry the inert
+--echo # "heap_test_guard" keyword: the one-shot injection removes its own
+--echo # keyword when it fires, and a keyword list that becomes empty while
+--echo # debugging is on matches every keyword.
+--echo #
+
+CALL mtr.add_suppression("Index for table .t1. is corrupt");
+
+--echo #
+--echo # The hash key k1 is moved to the new value, then the rb-tree node
+--echo # allocation for k2 fails once: the failure is reported as out of
+--echo # memory, not as a duplicate, and both keys are moved back
+--echo #
+
+CREATE TABLE t1 (
+  id INT,
+  a INT,
+  b INT,
+  KEY k1 (a),
+  KEY k2 (b) USING BTREE
+) ENGINE=HEAP;
+
+INSERT INTO t1 VALUES (1, 10, 100), (2, 20, 200);
+
+SET SESSION debug_dbug="+d,once_simulate_tree_insert_oom,heap_test_guard";
+--error ER_OUT_OF_RESOURCES
+UPDATE t1 SET a= 11, b= 101 WHERE id = 1;
+SET SESSION debug_dbug="";
+
+CHECK TABLE t1;
+--echo # Both indexes must still find the row under its unchanged values
+SELECT id, a, b FROM t1 FORCE INDEX(k1) WHERE a = 10;
+SELECT id, a, b FROM t1 FORCE INDEX(k2) WHERE b = 100;
+SELECT COUNT(*) FROM t1 FORCE INDEX(k1) WHERE a = 11;
+SELECT COUNT(*) FROM t1 FORCE INDEX(k2) WHERE b = 101;
+SELECT id, a, b FROM t1 ORDER BY id;
+
+--echo #
+--echo # A duplicate key is still reported as a duplicate
+--echo #
+
+CREATE TABLE t2 (
+  id INT,
+  a INT,
+  UNIQUE KEY k1 (a) USING BTREE
+) ENGINE=HEAP;
+
+INSERT INTO t2 VALUES (1, 10), (2, 20);
+
+--error ER_DUP_ENTRY
+UPDATE t2 SET a= 20 WHERE id = 1;
+
+CHECK TABLE t2;
+SELECT id, a FROM t2 ORDER BY id;
+DROP TABLE t2;
+
+--echo #
+--echo # The allocation failure persists, so restoring the old k2 entry
+--echo # fails as well: the already moved k1 no longer describes the
+--echo # record, the table is marked crashed, and every read and write on
+--echo # it is refused until it is emptied
+--echo #
+
+SET SESSION debug_dbug="+d,simulate_tree_insert_oom,heap_test_guard";
+--error ER_OUT_OF_RESOURCES
+UPDATE t1 SET a= 11, b= 101 WHERE id = 1;
+SET SESSION debug_dbug="";
+
+CHECK TABLE t1;
+--error ER_NOT_KEYFILE
+SELECT id, a, b FROM t1 ORDER BY id;
+--error ER_NOT_KEYFILE
+SELECT id, a, b FROM t1 FORCE INDEX(k1) WHERE a = 10;
+--error ER_NOT_KEYFILE
+INSERT INTO t1 VALUES (3, 30, 300);
+--error ER_NOT_KEYFILE
+UPDATE t1 SET a= 12 WHERE id = 2;
+--error ER_NOT_KEYFILE
+DELETE FROM t1 WHERE id = 2;
+
+--echo # Emptying the table rebuilds the indexes from nothing
+TRUNCATE TABLE t1;
+CHECK TABLE t1;
+INSERT INTO t1 VALUES (5, 50, 500);
+SELECT id, a, b FROM t1 FORCE INDEX(k2) WHERE b = 500;
+
+DROP TABLE t1;
diff --git a/mysys/tree.c b/mysys/tree.c
index f20752b732785..1e212c63fd332 100644
--- a/mysys/tree.c
+++ b/mysys/tree.c
@@ -106,6 +106,7 @@ void init_tree(TREE *tree, size_t default_alloc_size, size_t memory_limit,
   tree->custom_arg = custom_arg;
   tree->my_flags= my_flags;
   tree->flag= 0;
+  tree->error= TREE_ERROR_NONE;
   if (!free_element && size >= 0 &&
       ((uint) size <= sizeof(void*) || ((uint) size & (sizeof(void*)-1))))
   {
@@ -274,13 +275,35 @@ TREE_ELEMENT *tree_insert(TREE *tree, void *key, uint key_size,
     }
 
     key_size+=tree->size_of_element;
+    /*
+      Debug hooks that make this node allocation fail as if the allocator
+      had returned NULL.  "simulate_tree_insert_oom" fails every insert
+      until the caller disarms it; "once_simulate_tree_insert_oom" fails
+      only the next one and disarms itself.  Neither name may be a prefix
+      of the other: DBUG_SET() matches an existing list entry by prefix
+      and would merge into it instead of adding the keyword.
+    */
+    DBUG_EXECUTE_IF("once_simulate_tree_insert_oom",
+                    {
+                      DBUG_SET("-d,once_simulate_tree_insert_oom");
+                      tree->error= TREE_ERROR_OOM;
+                      return(NULL);
+                    });
+    DBUG_EXECUTE_IF("simulate_tree_insert_oom",
+                    {
+                      tree->error= TREE_ERROR_OOM;
+                      return(NULL);
+                    });
     if (tree->with_delete)
       element=(TREE_ELEMENT *) my_malloc(key_memory_TREE, alloc_size,
                                          MYF(tree->my_flags | MY_WME));
     else
       element=(TREE_ELEMENT *) alloc_root(&tree->mem_root,alloc_size);
     if (!element)
+    {
+      tree->error= TREE_ERROR_OOM;
       return(NULL);
+    }
     **parent=element;
     element->left=element->right= &null_element;
     if (!tree->offset_to_key)
@@ -303,7 +326,10 @@ TREE_ELEMENT *tree_insert(TREE *tree, void *key, uint key_size,
   else
   {
     if (tree->flag & TREE_NO_DUPS)
+    {
+      tree->error= TREE_ERROR_DUP_KEY;
       return(NULL);
+    }
     element->count++;
     /* Avoid a wrap over of the count. */
     if (! element->count)
diff --git a/storage/heap/CMakeLists.txt b/storage/heap/CMakeLists.txt
index d46669d5db757..8c6d955374097 100644
--- a/storage/heap/CMakeLists.txt
+++ b/storage/heap/CMakeLists.txt
@@ -32,7 +32,8 @@ IF(WITH_UNIT_TESTS)
   TARGET_LINK_LIBRARIES(hp_test1 heap mysys dbug strings)
   ADD_EXECUTABLE(hp_test2 hp_test2.c)
   TARGET_LINK_LIBRARIES(hp_test2 heap mysys dbug strings)
-  MY_ADD_TESTS(hp_test_hash hp_test_scan hp_test_freelist hp_test_concurrent LINK_LIBRARIES heap mysys dbug strings)
+  MY_ADD_TESTS(hp_test_hash hp_test_scan hp_test_freelist hp_test_concurrent
+               hp_test_update LINK_LIBRARIES heap mysys dbug strings)
 
   INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/sql
                       ${CMAKE_SOURCE_DIR}/include)
diff --git a/storage/heap/_check.c b/storage/heap/_check.c
index 9f408b73da203..6e7625f9d4601 100644
--- a/storage/heap/_check.c
+++ b/storage/heap/_check.c
@@ -51,6 +51,9 @@ int heap_check_heap(const HP_INFO *info, my_bool print_status)
   uchar *current_ptr= info->current_ptr;
   DBUG_ENTER("heap_check_heap");
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(1);                     /* Already known to be damaged */
+
   for (error=key= 0 ; key < share->keys ; key++)
   {
     if (share->keydef[key].algorithm == HA_KEY_ALG_BTREE)
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index 5a3fb72d56b2b..bab57ada339e2 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -528,7 +528,15 @@ int ha_heap::reset_auto_increment(ulonglong value)
 int ha_heap::external_lock(THD *thd, int lock_type)
 {
 #if !defined(DBUG_OFF) && defined(EXTRA_HEAP_DEBUG)
-  if (lock_type == F_UNLCK && file->s->changed && heap_check_heap(file, 0))
+  /*
+    A table already marked crashed is knowingly inconsistent; every data
+    access on it fails with HA_ERR_CRASHED, so re-detecting the damage
+    here would only raise a second error into a diagnostics area that
+    can already be OK (e.g. after UNLOCK TABLES) and fire the
+    Diagnostics_area assertion.
+  */
+  if (lock_type == F_UNLCK && file->s->changed &&
+      !heap_is_crashed(file->s) && heap_check_heap(file, 0))
     return HA_ERR_CRASHED;
 #endif
   if (lock_type == F_UNLCK)
diff --git a/storage/heap/hp_clear.c b/storage/heap/hp_clear.c
index 15a49b0e29a52..8db3da68387c1 100644
--- a/storage/heap/hp_clear.c
+++ b/storage/heap/hp_clear.c
@@ -43,6 +43,7 @@ void hp_clear(HP_SHARE *info)
   info->data_length= 0;
   info->blength=1;
   info->changed=0;
+  info->state_changed=0;
   info->del_link=0;
   info->key_version++;
   info->file_version++;
diff --git a/storage/heap/hp_delete.c b/storage/heap/hp_delete.c
index 5dbd3590c8c82..aa7656f428e58 100644
--- a/storage/heap/hp_delete.c
+++ b/storage/heap/hp_delete.c
@@ -146,6 +146,8 @@ int heap_delete(HP_INFO *info, const uchar *record)
   DBUG_ENTER("heap_delete");
   DBUG_PRINT("enter",("info: %p  record: %p", info, record));
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   test_active(info);
   hp_flush_pending_blob_free(info);
 
diff --git a/storage/heap/hp_rfirst.c b/storage/heap/hp_rfirst.c
index 903fd42a135ed..e97851cbdc088 100644
--- a/storage/heap/hp_rfirst.c
+++ b/storage/heap/hp_rfirst.c
@@ -24,6 +24,8 @@ int heap_rfirst(HP_INFO *info, uchar *record, int inx)
   HP_KEYDEF *keyinfo = share->keydef + inx;
   
   DBUG_ENTER("heap_rfirst");
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   info->lastinx= inx;
   info->key_version= info->s->key_version;
 
diff --git a/storage/heap/hp_rkey.c b/storage/heap/hp_rkey.c
index 8a9b515d8af04..f0abee4ae9a8a 100644
--- a/storage/heap/hp_rkey.c
+++ b/storage/heap/hp_rkey.c
@@ -25,6 +25,8 @@ int heap_rkey(HP_INFO *info, uchar *record, int inx, const uchar *key,
   DBUG_ENTER("heap_rkey");
   DBUG_PRINT("enter",("info: %p  inx: %d", info, inx));
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   if ((uint) inx >= share->keys)
   {
     DBUG_RETURN(my_errno= HA_ERR_WRONG_INDEX);
diff --git a/storage/heap/hp_rlast.c b/storage/heap/hp_rlast.c
index 5b31bfccf07c0..6f68600bafd5d 100644
--- a/storage/heap/hp_rlast.c
+++ b/storage/heap/hp_rlast.c
@@ -25,6 +25,8 @@ int heap_rlast(HP_INFO *info, uchar *record, int inx)
   HP_KEYDEF *keyinfo= share->keydef + inx;
 
   DBUG_ENTER("heap_rlast");
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   info->lastinx= inx;
   info->key_version= info->s->key_version;
   if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
diff --git a/storage/heap/hp_rnext.c b/storage/heap/hp_rnext.c
index 774731624fd96..f10d1ff7fe1f1 100644
--- a/storage/heap/hp_rnext.c
+++ b/storage/heap/hp_rnext.c
@@ -24,7 +24,9 @@ int heap_rnext(HP_INFO *info, uchar *record)
   HP_SHARE *share=info->s;
   HP_KEYDEF *keyinfo;
   DBUG_ENTER("heap_rnext");
-  
+
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   if (info->lastinx < 0)
     DBUG_RETURN(my_errno=HA_ERR_WRONG_INDEX);
 
diff --git a/storage/heap/hp_rprev.c b/storage/heap/hp_rprev.c
index 948d1db15ec53..feda190da4e06 100644
--- a/storage/heap/hp_rprev.c
+++ b/storage/heap/hp_rprev.c
@@ -26,6 +26,8 @@ int heap_rprev(HP_INFO *info, uchar *record)
   HP_KEYDEF *keyinfo;
   DBUG_ENTER("heap_rprev");
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   if (info->lastinx < 0)
     DBUG_RETURN(my_errno=HA_ERR_WRONG_INDEX);
   keyinfo = share->keydef + info->lastinx;
diff --git a/storage/heap/hp_rrnd.c b/storage/heap/hp_rrnd.c
index 045804a94afe7..7ce640ac03477 100644
--- a/storage/heap/hp_rrnd.c
+++ b/storage/heap/hp_rrnd.c
@@ -31,6 +31,8 @@ int heap_rrnd(register HP_INFO *info, uchar *record, uchar *pos)
   DBUG_ENTER("heap_rrnd");
   DBUG_PRINT("enter",("info: %p  pos: %p", info, pos));
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   info->lastinx= -1;
   if (!(info->current_ptr= pos))
   {
diff --git a/storage/heap/hp_rsame.c b/storage/heap/hp_rsame.c
index 1ab2511d617ba..cbc373e410404 100644
--- a/storage/heap/hp_rsame.c
+++ b/storage/heap/hp_rsame.c
@@ -31,6 +31,8 @@ int heap_rsame(register HP_INFO *info, uchar *record, int inx)
   HP_SHARE *share=info->s;
   DBUG_ENTER("heap_rsame");
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   test_active(info);
   if (info->current_ptr[share->visible])
   {
diff --git a/storage/heap/hp_scan.c b/storage/heap/hp_scan.c
index 12347c9bf01a8..caa4914bd4de9 100644
--- a/storage/heap/hp_scan.c
+++ b/storage/heap/hp_scan.c
@@ -28,6 +28,8 @@
 int heap_scan_init(register HP_INFO *info)
 {
   DBUG_ENTER("heap_scan_init");
+  if (heap_is_crashed(info->s))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   info->lastinx= -1;
   info->current_record= (ulong) ~0L;		/* No current record */
   info->update=0;
@@ -43,6 +45,8 @@ int heap_scan(register HP_INFO *info, uchar *record)
   ulong pos;
   DBUG_ENTER("heap_scan");
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   /*
     Scan boundary: total_records + deleted == block.last_allocated.
 
diff --git a/storage/heap/hp_test_update-t.c b/storage/heap/hp_test_update-t.c
new file mode 100644
index 0000000000000..27daf7cf7e2a7
--- /dev/null
+++ b/storage/heap/hp_test_update-t.c
@@ -0,0 +1,360 @@
+/*
+   Unit tests for the key recovery of heap_update().
+
+   heap_update() moves every changed key entry to its new key value before it
+   updates the record itself, so a failure raised afterwards leaves the index
+   describing a record image that is not in the table.  The recovery at the
+   err: label moves those entries back for the errors it knows how to recover
+   from, and marks the table crashed when it cannot restore a key, so that no
+   later statement can read from or write to an index that no longer
+   describes the data.
+
+   The failure used here - an rb-tree node allocation failure - is injected
+   with DBUG keywords inside tree_insert(), so these tests only run on debug
+   builds.  The keyword lists always carry the inert "heap_test_guard"
+   keyword: a keyword list that becomes empty while debugging is on matches
+   every keyword and would turn on all of the debug output.
+*/
+
+#include <my_global.h>
+#include <my_sys.h>
+#include <m_string.h>
+#include <tap.h>
+#include "heap.h"
+#include "heapdef.h"
+
+/*
+  Record layout: (null bitmap, int4 a, int4 b)
+    byte 0:     null bitmap
+    bytes 1-4:  a
+    bytes 5-8:  b
+*/
+#define REC_LENGTH  9
+#define A_OFFSET    1
+#define B_OFFSET    5
+
+/* Never a valid errkey value, so it shows whether errkey was written to */
+#define ERRKEY_UNSET (-2)
+
+/*
+  Everything below is driven by the injected failures, so a build without
+  DBUG has nothing to call it with.  The tests and the helpers only they use
+  are left out of such a build entirely, rather than only left uncalled,
+  which would warn about every one of them.
+*/
+#ifndef DBUG_OFF
+
+static void build_rec(uchar *rec, int32 a, int32 b)
+{
+  memset(rec, 0, REC_LENGTH);
+  int4store(rec + A_OFFSET, a);
+  int4store(rec + B_OFFSET, b);
+}
+
+
+/*
+  Create a table with 'keys' keys: key 0 over column a and key 1 over
+  column b.  Only key 0 can be made unique; the tests that need a
+  duplicate use a single key.
+*/
+
+static int create_and_open(const char *name, uint keys,
+                           enum ha_key_alg alg_a, enum ha_key_alg alg_b,
+                           uint flag_a, HP_SHARE **share, HP_INFO **info)
+{
+  HP_KEYDEF keydef[2];
+  HA_KEYSEG keyseg[2];
+  HP_CREATE_INFO ci;
+  my_bool unused;
+  uint i;
+
+  memset(keyseg, 0, sizeof(keyseg));
+  memset(keydef, 0, sizeof(keydef));
+  for (i= 0; i < 2; i++)
+  {
+    keyseg[i].type=    HA_KEYTYPE_BINARY;
+    keyseg[i].start=   i ? B_OFFSET : A_OFFSET;
+    keyseg[i].length=  4;
+    keyseg[i].charset= &my_charset_bin;
+
+    keydef[i].keysegs= 1;
+    keydef[i].seg=     &keyseg[i];
+    keydef[i].length=  4;
+  }
+  keydef[0].algorithm= alg_a;
+  keydef[0].flag=      flag_a;
+  keydef[1].algorithm= alg_b;
+
+  memset(&ci, 0, sizeof(ci));
+  ci.keys=           keys;
+  ci.keydef=         keydef;
+  ci.reclength=      REC_LENGTH;
+  ci.max_records=    1000;
+  ci.min_records=    10;
+  ci.max_table_size= 1024 * 1024;
+
+  if (heap_create(name, &ci, share, &unused))
+    return 1;
+  if (!(*info= heap_open(name, 2)))
+    return 1;
+  heap_extra(*info, HA_EXTRA_NO_READCHECK);
+  return 0;
+}
+
+
+static int position_on(HP_INFO *info, int keynr, int32 value, uchar *rec)
+{
+  uchar key[4];
+  int4store(key, value);
+  return heap_rkey(info, rec, keynr, key, (key_part_map) 1, HA_READ_KEY_EXACT);
+}
+
+
+static my_bool row_found(HP_INFO *info, int keynr, int32 value)
+{
+  uchar rec[REC_LENGTH];
+  return position_on(info, keynr, value, rec) == 0;
+}
+
+
+/*
+  Insert (1, 10) and (2, 20) and position on the first one through key 0.
+*/
+
+static int fill_and_position(HP_INFO *info, uchar *rec)
+{
+  build_rec(rec, 1, 10);
+  if (heap_write(info, rec))
+    return 1;
+  build_rec(rec, 2, 20);
+  if (heap_write(info, rec))
+    return 1;
+  build_rec(rec, 1, 10);
+  return position_on(info, 0, 1, rec) != 0;
+}
+
+
+/*
+  Test 1: a failed rb-tree node allocation is reported as such.
+
+  The key is not unique, so a duplicate cannot even occur, yet before the
+  fix the failure was reported as HA_ERR_FOUND_DUPP_KEY - with a fabricated
+  duplicate value and a key that has no uniqueness to violate.  The failure
+  is injected only once, so restoring the old key succeeds and the table
+  stays usable.
+*/
+
+static void test_rb_alloc_failure_is_oom(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], new_rec[REC_LENGTH];
+  ulonglong index_length_before;
+  int error;
+
+  if (create_and_open("test_upd_oom", 1, HA_KEY_ALG_BTREE, HA_KEY_ALG_BTREE,
+                      0, &share, &info) ||
+      fill_and_position(info, rec))
+  {
+    ok(0, "setup failed: %d", my_errno);
+    skip(5, "setup failed");
+    return;
+  }
+
+  index_length_before= share->index_length;
+  build_rec(new_rec, 11, 10);
+  info->errkey= ERRKEY_UNSET;
+
+  DBUG_PUSH("d,once_simulate_tree_insert_oom,heap_test_guard");
+  error= heap_update(info, rec, new_rec);
+  DBUG_POP();
+
+  ok(error == HA_ERR_OUT_OF_MEM,
+     "allocation failure reported as out of memory, not as a duplicate "
+     "(got %d)", error);
+  ok(info->errkey == -1,
+     "errkey says no key error for a non-duplicate failure (got %d)",
+     info->errkey);
+  ok(share->index_length == index_length_before,
+     "index size unchanged after the recovery");
+  ok(!heap_is_crashed(share), "table not crashed: the recovery succeeded");
+  ok(heap_check_heap(info, 0) == 0, "index consistent after the recovery");
+  ok(row_found(info, 0, 1), "row still found through its old key value");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
+/*
+  Test 2: a real duplicate is still reported as a duplicate.
+
+  Guards the classification above against reporting every rb-tree insert
+  failure as an allocation failure.
+*/
+
+static void test_rb_duplicate_still_reported(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], new_rec[REC_LENGTH];
+  int error;
+
+  if (create_and_open("test_upd_dupp", 1, HA_KEY_ALG_BTREE, HA_KEY_ALG_BTREE,
+                      HA_NOSAME, &share, &info) ||
+      fill_and_position(info, rec))
+  {
+    ok(0, "setup failed: %d", my_errno);
+    skip(3, "setup failed");
+    return;
+  }
+
+  /* a=2 already exists on the second row */
+  build_rec(new_rec, 2, 10);
+  info->errkey= ERRKEY_UNSET;
+
+  error= heap_update(info, rec, new_rec);
+
+  ok(error == HA_ERR_FOUND_DUPP_KEY,
+     "duplicate key still reported as a duplicate (got %d)", error);
+  ok(info->errkey == 0, "errkey names the duplicate key (got %d)",
+     info->errkey);
+  ok(heap_check_heap(info, 0) == 0, "index consistent after the recovery");
+  ok(row_found(info, 0, 1), "row still found through its old key value");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
+/*
+  Test 3: a key that was already moved is moved back.
+
+  Key 0 (hash) is moved to its new value before key 1 (rb-tree) fails, so
+  the recovery has to move key 0 back to the value the record still holds.
+*/
+
+static void test_earlier_key_is_moved_back(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], new_rec[REC_LENGTH];
+  int error;
+
+  if (create_and_open("test_upd_earlier", 2, HA_KEY_ALG_HASH,
+                      HA_KEY_ALG_BTREE, 0, &share, &info) ||
+      fill_and_position(info, rec))
+  {
+    ok(0, "setup failed: %d", my_errno);
+    skip(4, "setup failed");
+    return;
+  }
+
+  build_rec(new_rec, 11, 110);
+
+  DBUG_PUSH("d,once_simulate_tree_insert_oom,heap_test_guard");
+  error= heap_update(info, rec, new_rec);
+  DBUG_POP();
+
+  ok(error == HA_ERR_OUT_OF_MEM, "update failed with out of memory (got %d)",
+     error);
+  ok(!heap_is_crashed(share), "table not crashed: the recovery succeeded");
+  ok(heap_check_heap(info, 0) == 0, "both indexes consistent after recovery");
+  ok(row_found(info, 0, 1), "hash key moved back to the old value");
+  ok(row_found(info, 1, 10), "rb-tree key holds the old value");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
+/*
+  Test 4: a failed recovery marks the table crashed and shuts it down.
+
+  The allocation failure is left switched on, so restoring the old key 1
+  (rb-tree) entry fails as well.  The already moved key 0 (hash) then no
+  longer describes the record, so the table is marked crashed: every read
+  and write on it fails with HA_ERR_CRASHED and heap_check_heap() reports
+  it as damaged, until the table is emptied, which rebuilds the indexes
+  from nothing.
+*/
+
+static void test_failed_recovery_marks_crashed(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], new_rec[REC_LENGTH];
+  int error;
+
+  if (create_and_open("test_upd_crashed", 2, HA_KEY_ALG_HASH,
+                      HA_KEY_ALG_BTREE, 0, &share, &info) ||
+      fill_and_position(info, rec))
+  {
+    ok(0, "setup failed: %d", my_errno);
+    skip(8, "setup failed");
+    return;
+  }
+
+  build_rec(new_rec, 11, 110);
+
+  DBUG_PUSH("d,simulate_tree_insert_oom,heap_test_guard");
+  error= heap_update(info, rec, new_rec);
+  DBUG_POP();
+
+  ok(error == HA_ERR_OUT_OF_MEM, "update failed with out of memory (got %d)",
+     error);
+  ok(heap_is_crashed(share),
+     "table marked crashed: one key could not be restored");
+  ok(heap_check_heap(info, 0) != 0, "heap_check_heap() reports the damage");
+
+  build_rec(rec, 1, 10);
+  build_rec(new_rec, 3, 30);
+  ok(heap_update(info, rec, new_rec) == HA_ERR_CRASHED,
+     "update on a crashed table is refused");
+  ok(!row_found(info, 0, 1) && my_errno == HA_ERR_CRASHED,
+     "key read on a crashed table is refused");
+  build_rec(rec, 4, 40);
+  ok(heap_write(info, rec) == HA_ERR_CRASHED,
+     "write on a crashed table is refused");
+  ok(heap_scan_init(info) == HA_ERR_CRASHED,
+     "scan on a crashed table is refused");
+
+  heap_clear(info);
+  ok(!heap_is_crashed(share), "emptying the table clears the crashed state");
+  build_rec(rec, 5, 50);
+  ok(heap_write(info, rec) == 0 && row_found(info, 0, 5),
+     "the emptied table accepts and finds rows again");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+#endif /* !DBUG_OFF */
+
+
+int main(int argc __attribute__((unused)),
+         char **argv __attribute__((unused)))
+{
+  MY_INIT("hp_test_update");
+
+#ifdef DBUG_OFF
+  skip_all("the failures under test are injected with DBUG keywords");
+#else
+  plan(24);
+
+  diag("Test 1: rb-tree allocation failure is not a duplicate key");
+  test_rb_alloc_failure_is_oom();
+
+  diag("Test 2: a real duplicate is still reported as a duplicate");
+  test_rb_duplicate_still_reported();
+
+  diag("Test 3: an already moved key is moved back");
+  test_earlier_key_is_moved_back();
+
+  diag("Test 4: a failed recovery marks the table crashed");
+  test_failed_recovery_marks_crashed();
+#endif
+
+  my_end(0);
+  return exit_status();
+}
diff --git a/storage/heap/hp_update.c b/storage/heap/hp_update.c
index f822c8daaf803..3e308a729fbcc 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -20,13 +20,15 @@
 
 int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
 {
-  HP_KEYDEF *keydef, *end, *p_lastinx;
+  HP_KEYDEF *keydef, *keydef_end, *p_lastinx;
   uchar *pos, *recovery_ptr;
   struct st_hp_hash_info *recovery_hash_ptr;
   my_bool auto_key_changed= 0, key_changed= 0;
   HP_SHARE *share= info->s;
   DBUG_ENTER("heap_update");
 
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
   test_active(info);
   hp_flush_pending_blob_free(info);
   pos=info->current_ptr;
@@ -41,7 +43,9 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
   recovery_hash_ptr= info->current_hash_ptr;
 
   p_lastinx= share->keydef + info->lastinx;
-  for (keydef= share->keydef, end= keydef + share->keys; keydef < end; keydef++)
+  for (keydef= share->keydef, keydef_end= keydef + share->keys;
+       keydef < keydef_end;
+       keydef++)
   {
     if (hp_rec_key_cmp(keydef, heap_new, old, NULL))
     {
@@ -253,32 +257,67 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
   DBUG_RETURN(0);
 
  err:
-  if (my_errno == HA_ERR_FOUND_DUPP_KEY)
+  info->errkey= -1;                             /* If not a key error */
+  if (my_errno == HA_ERR_FOUND_DUPP_KEY || my_errno == HA_ERR_OUT_OF_MEM ||
+      my_errno == ENOMEM || my_errno == HA_ERR_RECORD_FILE_FULL)
   {
-    info->errkey = (int) (keydef - share->keydef);
-    if (keydef->algorithm == HA_KEY_ALG_BTREE)
+    int org_error= my_errno;
+    if (keydef == keydef_end)
     {
-      /* we don't need to delete non-inserted key from rb-tree */
-      if ((*keydef->write_key)(info, keydef, old, pos))
+      /*
+        We got a failure after all key parts have been written, like
+        while writing blobs. Alternatively there are no keys.
+        Decrement keydef so that it points at the last key part or
+        before share->keydef.
+      */
+      DBUG_ASSERT(share->keydef);
+      keydef--;
+    }
+    else
+    {
+      /* keydef points to the last key that failed */
+      if (org_error == HA_ERR_FOUND_DUPP_KEY)
+        info->errkey = (int) (keydef - share->keydef);
+      if (keydef->algorithm == HA_KEY_ALG_BTREE)
       {
-        if (++(share->records) == share->blength)
-	  share->blength+= share->blength;
-        DBUG_RETURN(my_errno);
+        /* we don't need to delete non-inserted key from rb-tree */
+        if ((*keydef->write_key)(info, keydef, old, pos))
+        {
+          heap_mark_crashed(share);
+          if (++(share->records) == share->blength)
+            share->blength+= share->blength;
+          DBUG_RETURN(my_errno);
+        }
+        keydef--;
       }
-      keydef--;
     }
+    /* Restore all modified keys */
     while (keydef >= share->keydef)
     {
       if (hp_rec_key_cmp(keydef, heap_new, old, NULL))
       {
 	if ((*keydef->delete_key)(info, keydef, heap_new, pos, 0) ||
 	    (*keydef->write_key)(info, keydef, old, pos))
+	{
+	  heap_mark_crashed(share);
 	  break;
+	}
       }
       keydef--;
     }
     info->current_ptr= recovery_ptr;
     info->current_hash_ptr= recovery_hash_ptr;
+    my_errno= org_error;
+  }
+  else
+  {
+    /*
+      An error we do not know how to recover from.  The changed keys have
+      been moved to the new values, or could not be removed from them,
+      while the record keeps the old ones, so the index no longer
+      describes the data.
+    */
+    heap_mark_crashed(share);
   }
   if (++(share->records) == share->blength)
     share->blength+= share->blength;
diff --git a/storage/heap/hp_write.c b/storage/heap/hp_write.c
index 5c30caef4c633..8741e941184e9 100644
--- a/storage/heap/hp_write.c
+++ b/storage/heap/hp_write.c
@@ -35,6 +35,8 @@ int heap_write(HP_INFO *info, const uchar *record)
   uchar *pos;
   HP_SHARE *share=info->s;
   DBUG_ENTER("heap_write");
+  if (heap_is_crashed(share))
+    DBUG_RETURN(my_errno= HA_ERR_CRASHED);
 #ifndef DBUG_OFF
   if (info->mode & O_RDONLY)
   {
@@ -123,11 +125,12 @@ int heap_write(HP_INFO *info, const uchar *record)
   Write a key to rb_tree-index 
 */
 
-int hp_rb_write_key(HP_INFO *info, HP_KEYDEF *keyinfo, const uchar *record, 
+int hp_rb_write_key(HP_INFO *info, HP_KEYDEF *keyinfo, const uchar *record,
 		    uchar *recpos)
 {
   heap_rb_param custom_arg;
   size_t old_allocated;
+  TREE_ELEMENT *element;
 
   custom_arg.keyseg= keyinfo->seg;
   custom_arg.key_length= hp_rb_make_key(keyinfo, info->recbuf, record, recpos);
@@ -142,10 +145,16 @@ int hp_rb_write_key(HP_INFO *info, HP_KEYDEF *keyinfo, const uchar *record,
     keyinfo->rb_tree.flag= 0;
   }
   old_allocated= keyinfo->rb_tree.allocated;
-  if (!tree_insert(&keyinfo->rb_tree, (void*)info->recbuf,
-		   custom_arg.key_length, &custom_arg))
+  element= tree_insert(&keyinfo->rb_tree, (void*) info->recbuf,
+                       custom_arg.key_length, &custom_arg);
+  if (!element)
   {
-    my_errno= HA_ERR_FOUND_DUPP_KEY;
+    if (keyinfo->rb_tree.error == TREE_ERROR_OOM)
+      my_errno= HA_ERR_OUT_OF_MEM;
+    else if (keyinfo->rb_tree.error == TREE_ERROR_DUP_KEY)
+      my_errno= HA_ERR_FOUND_DUPP_KEY;
+    else
+      my_errno= HA_ERR_INTERNAL_ERROR;
     return 1;
   }
   info->s->index_length+= (keyinfo->rb_tree.allocated-old_allocated);