From 9482ccc36704c79f674097ce50ba48897eaee581 Mon Sep 17 00:00:00 2001
From: Arcadiy Ivanov <arcadiy@ivanov.biz>
Date: Tue, 28 Jul 2026 16:05:52 -0400
Subject: [PATCH 1/2] MDEV-40378 heap: roll back key changes when a blob write
 fails in `heap_update()`

`heap_update()` moves all changed key entries to the new key values
**before** writing the new blob chains. When a blob chain write then
failed (e.g. with `HA_ERR_RECORD_FILE_FULL`), the rollback restored the
record bytes and blob chain pointers, but the `err:` label only undid
key changes for `HA_ERR_FOUND_DUPP_KEY` -- historically the only
possible failure once the key loop had run. The hash/btree entries were
left keyed on the new values while pointing at a record holding the old
values, corrupting the index:

1. index lookups by the old key value missed the row
2. `CHECK TABLE` reported the table corrupt
3. on debug builds the heap consistency check in
   `ha_heap::external_lock()` raised a second error into an already-set
   diagnostics area, firing a `Diagnostics_area` assertion on the next
   statement

Fix: widen the `err:` recovery to run for `HA_ERR_RECORD_FILE_FULL`,
`HA_ERR_OUT_OF_MEM` and `ENOMEM` as well, so a failure raised after the
key loop also moves every changed key back to its old value. One
recovery path now serves every failure that leaves keys moved to their
new values, including any future error source in the key loop itself.

The `err:` block assumed the failure happened **inside** the key loop,
so that `keydef` addresses the partially processed keydef. A blob-chain
write fails after that loop has run to completion, and therefore
arrives with `keydef == keydef_end`. Reading `info->errkey` and
`keydef->algorithm` from there addresses `share->keydef[share->keys]`;
as `sizeof(HP_KEYDEF)` (888) far exceeds the key segments and blob
descriptors that follow the keydef array, that read runs past the end
of the `HP_SHARE` allocation, and the rollback sweep then dereferences
a garbage `keydef->seg` in `hp_rec_key_cmp()`. So the recovery
distinguishes the two failure sites: with `keydef == keydef_end` there
is no partly updated key to repair and none to name in `info->errkey`,
and the sweep starts at the last keydef instead. The same branch also
covers a table with no keys at all (`share->keys == 0`), where the
sweep has nothing to do.

`info->errkey` is initialized to `-1` on entry to `err:`, so a failure
that is not a key error can never expose a stale key number from an
earlier operation.

The original errno is captured before the recovery and restored after
it, so that a rollback `write_key` failure (which `hp_rb_write_key()`
reports as `HA_ERR_FOUND_DUPP_KEY` with a stale `errkey`) cannot mask
it.

The new test `heap.blob_update_key_rollback` exercises hash, BTREE, two
changed indexes, an index on an unchanged column (which the rollback
must leave untouched), a partial multi-row UPDATE, and a table with no
indexes at all; each asserts the table stays consistent after the
failure via `CHECK TABLE` and index lookups.
---
 .../heap/blob_update_key_rollback.result      | 186 ++++++++++++++++++
 .../suite/heap/blob_update_key_rollback.test  | 179 +++++++++++++++++
 storage/heap/hp_update.c                      |  44 +++--
 3 files changed, 398 insertions(+), 11 deletions(-)
 create mode 100644 mysql-test/suite/heap/blob_update_key_rollback.result
 create mode 100644 mysql-test/suite/heap/blob_update_key_rollback.test

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/storage/heap/hp_update.c b/storage/heap/hp_update.c
index f822c8daaf803..aa40524649497 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -20,7 +20,7 @@
 
 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;
@@ -41,7 +41,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,20 +255,39 @@ 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 */
+      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))
+        {
+          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))
@@ -279,6 +300,7 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
     }
     info->current_ptr= recovery_ptr;
     info->current_hash_ptr= recovery_hash_ptr;
+    my_errno= org_error;
   }
   if (++(share->records) == share->blength)
     share->blength+= share->blength;

From f675abcea9570e51cf3a1ddee789cc630958caba Mon Sep 17 00:00:00 2001
From: Arcadiy Ivanov <arcadiy@ivanov.biz>
Date: Tue, 28 Jul 2026 20:54:27 -0400
Subject: [PATCH 2/2] MDEV-40431 heap: classify rb-tree insert failures, mark
 the table crashed when key recovery fails

`hp_rb_write_key()` reported **every** rejected `tree_insert()` as
`HA_ERR_FOUND_DUPP_KEY`, although `tree_insert()` also returns NULL when
the allocation of the tree node fails. An out of memory during a
BTREE-indexed UPDATE was therefore reported as a duplicate: the user got
`ER_DUP_ENTRY` with a fabricated value, possibly naming a key that is
not unique at all, `handler::is_fatal_error()` treated the allocation
failure as not fatal for callers asking for `HA_CHECK_DUP_KEY`, and the
`HA_ERR_OUT_OF_MEM` / `ENOMEM` arms of the `heap_update()` recovery list
were unreachable on the rb-tree path.

`tree_insert()` now records why it returned NULL in the new
`TREE::error` (`TREE_ERROR_OOM` or `TREE_ERROR_DUP_KEY`), and
`hp_rb_write_key()` maps that to `HA_ERR_OUT_OF_MEM` or
`HA_ERR_FOUND_DUPP_KEY` (`HA_ERR_INTERNAL_ERROR` defensively, should a
new NULL source appear). With the classification corrected, the
allocation failure lands in the `HA_ERR_OUT_OF_MEM` arm of the
`heap_update()` recovery, which moves the already changed keys back, so
correcting the error report does not trade the wrong message for a
silently corrupt index.

The recovery at `err:` keeps its explicit list of error codes: those are
the errors we know how to recover from, and recovering from errors whose
meaning we do not know is worse than stopping. What changes around it:

1. `info->errkey` is only set for `HA_ERR_FOUND_DUPP_KEY`, the single
   error it describes; every other failure leaves the `-1` that `err:`
   starts with.
2. When the recovery itself cannot restore a key -- the re-insert of an
   old key value fails -- or the failure is outside the list, the index
   no longer describes the data and the table is marked **crashed** in
   the new `HP_SHARE::state_changed` (bits and macros modeled on the
   `state.changed` of Maria, see `storage/maria/maria_def.h`). A crashed
   table refuses every read and write with `HA_ERR_CRASHED` ("Index for
   table is corrupt"), `heap_check_heap()` reports it as damaged, and
   `hp_clear()` -- reached through TRUNCATE or DELETE without WHERE --
   clears the state along with the data, because it rebuilds the indexes
   from nothing. Refusing a table known to be corrupt beats continuing
   and delivering wrong results.
3. `delete_key()` is assumed to succeed: it allocates no memory, so it
   cannot fail unless the table is already inconsistent, and building
   recovery logic and injected failures for that case would complicate
   the code for a scenario that cannot happen on a healthy table. If it
   ever does fail, the error falls outside the recovery list and the
   table is marked crashed, preserving the damaged state for analysis.

The debug-build consistency check in `ha_heap::external_lock(F_UNLCK)`
skips tables already marked crashed: their inconsistency is known and
deliberate, and re-detecting it would raise a second error into a
diagnostics area that can already hold OK, firing the
`Diagnostics_area` assertion the check exists to prevent.

The allocation failure is injected inside `tree_insert()` itself:
`simulate_tree_insert_oom` fails every node allocation until the caller
disarms it, `once_simulate_tree_insert_oom` fails only the next one and
disarms itself. The two names must not be a prefix of one another,
because `DBUG_SET()` matches an existing keyword by prefix and would
silently merge instead of adding. Callers arm the keywords themselves
and keep an inert guard keyword in the list, because a keyword list
that becomes empty while debugging is on matches every keyword.

The unit test `hp_test_update` drives the failures through the engine
API: the classification and its duplicate-key counterpart, an already
moved key being moved back, and the crashed lifecycle -- marking,
refusal of reads and writes, and the reset on emptying. With the
defects reintroduced, 8 of its 24 assertions fail.
`heap.update_key_rollback` covers the same from SQL; without the fix its
first UPDATE reports `ER_DUP_ENTRY 'Duplicate entry 101 for key k2'` on
a key that is not unique.
---
 include/heap.h                                |  12 +
 include/my_tree.h                             |   3 +
 .../suite/heap/update_key_rollback.result     | 105 +++++
 .../suite/heap/update_key_rollback.test       |  99 +++++
 mysys/tree.c                                  |  26 ++
 storage/heap/CMakeLists.txt                   |   3 +-
 storage/heap/_check.c                         |   3 +
 storage/heap/ha_heap.cc                       |  10 +-
 storage/heap/hp_clear.c                       |   1 +
 storage/heap/hp_delete.c                      |   2 +
 storage/heap/hp_rfirst.c                      |   2 +
 storage/heap/hp_rkey.c                        |   2 +
 storage/heap/hp_rlast.c                       |   2 +
 storage/heap/hp_rnext.c                       |   4 +-
 storage/heap/hp_rprev.c                       |   2 +
 storage/heap/hp_rrnd.c                        |   2 +
 storage/heap/hp_rsame.c                       |   2 +
 storage/heap/hp_scan.c                        |   4 +
 storage/heap/hp_test_update-t.c               | 360 ++++++++++++++++++
 storage/heap/hp_update.c                      |  19 +-
 storage/heap/hp_write.c                       |  17 +-
 21 files changed, 672 insertions(+), 8 deletions(-)
 create mode 100644 mysql-test/suite/heap/update_key_rollback.result
 create mode 100644 mysql-test/suite/heap/update_key_rollback.test
 create mode 100644 storage/heap/hp_test_update-t.c

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/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 aa40524649497..3e308a729fbcc 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -27,6 +27,8 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
   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;
@@ -274,12 +276,14 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
     else
     {
       /* keydef points to the last key that failed */
-      info->errkey = (int) (keydef - share->keydef);
+      if (org_error == HA_ERR_FOUND_DUPP_KEY)
+        info->errkey = (int) (keydef - share->keydef);
       if (keydef->algorithm == HA_KEY_ALG_BTREE)
       {
         /* 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);
@@ -294,7 +298,10 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
       {
 	if ((*keydef->delete_key)(info, keydef, heap_new, pos, 0) ||
 	    (*keydef->write_key)(info, keydef, old, pos))
+	{
+	  heap_mark_crashed(share);
 	  break;
+	}
       }
       keydef--;
     }
@@ -302,6 +309,16 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
     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;
   DBUG_RETURN(my_errno);
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);