diff --git a/mysql-test/suite/rpl/r/rpl_heartbeat_log_event_underflow.result b/mysql-test/suite/rpl/r/rpl_heartbeat_log_event_underflow.result
new file mode 100644
index 0000000000000..52b83915efd6e
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_heartbeat_log_event_underflow.result
@@ -0,0 +1,76 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+call mtr.add_suppression("Unexpected master's heartbeat data");
+call mtr.add_suppression("Event from master is shorter than its common header");
+call mtr.add_suppression("Relay log write failure: could not queue event from master");
+#
+# Initialize test data
+#
+connection master;
+create table t1 (a int);
+insert into t1 values (1);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/stop_slave.inc
+CHANGE MASTER TO MASTER_HEARTBEAT_PERIOD= 0.5;
+include/start_slave.inc
+connection master;
+set @saved_dbug= @@global.debug_dbug;
+#
+# Test Case 1: a heartbeat that leaves no room for a file name
+#
+connection master;
+set @@global.debug_dbug= "+d,binlog_sender_truncated_heartbeat";
+connection slave;
+include/stop_slave.inc
+include/start_slave.inc
+# Waiting for the IO thread to reject the truncated heartbeat
+include/wait_for_slave_io_error.inc [errno=1595]
+Last_IO_Error = 'Relay log write failure: could not queue event from master'
+# Ensure heartbeat validation rejected the event
+include/assert_grep.inc [The IO thread reported an inconsistent heartbeat event]
+include/assert_grep.inc [The heartbeat error should report an empty log_file_name]
+connection master;
+set @@global.debug_dbug= @saved_dbug;
+connection slave;
+include/stop_slave_sql.inc
+include/start_slave.inc
+#
+# Test Case 2: a heartbeat shorter than the common header
+#
+connection master;
+set @@global.debug_dbug= "+d,binlog_sender_undersized_heartbeat";
+connection slave;
+include/stop_slave.inc
+include/start_slave.inc
+# Waiting for the IO thread to reject the undersized heartbeat
+include/wait_for_slave_io_error.inc [errno=1595]
+Last_IO_Error = 'Relay log write failure: could not queue event from master'
+# Ensure the event was rejected before its header was read
+include/assert_grep.inc [The IO thread reported an event shorter than the common header]
+connection master;
+set @@global.debug_dbug= @saved_dbug;
+connection slave;
+include/stop_slave_sql.inc
+include/start_slave.inc
+#
+# Ensure replication works after the invalid heartbeat events
+#
+connection master;
+insert into t1 values (2);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/diff_tables.inc [master:t1,slave:t1]
+#
+# Cleanup
+#
+connection master;
+drop table t1;
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/rpl_end.inc
+# End of rpl_heartbeat_log_event_underflow.test
diff --git a/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow-master.opt b/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow-master.opt
new file mode 100644
index 0000000000000..a6e99a9fd5a93
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow-master.opt
@@ -0,0 +1 @@
+--binlog-checksum=CRC32
diff --git a/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow.test b/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow.test
new file mode 100644
index 0000000000000..2dc188c22c3a5
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_heartbeat_log_event_underflow.test
@@ -0,0 +1,147 @@
+#
+#   This test verifies that a replica rejects a heartbeat event that is
+# too short to carry the log file name it must hold. Two lengths are
+# covered. An event that reaches the heartbeat parser but leaves no room
+# for a file name must stop the IO thread with an error reporting an
+# empty file name. An event shorter than the common header must be
+# rejected before any parser reads that header. Replication must be able
+# to resume after each.
+#
+#   The debug injections send heartbeats of 17 and 13 bytes. MDEV-40643
+# reported that the 17-byte case underflowed the replica's computed name
+# length and crashed the replica.
+#
+#   Note rpl_heartbeat_log_event_underflow-master.opt sets
+# binlog_checksum=CRC32 on the master for Test Case 1. That is, those four
+# checksum bytes carry the 17-byte heartbeat over queue_event()'s 19-byte
+# minimum length.
+#
+# References:
+#   * MDEV-40643: Corrupt Heartbeat Log Event can Crash Slave
+#   * Bug#39377010: Improve malformed heartbeat handling
+#
+
+--source include/have_debug.inc
+# Test is format independent
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+--connection slave
+call mtr.add_suppression("Unexpected master's heartbeat data");
+call mtr.add_suppression("Event from master is shorter than its common header");
+call mtr.add_suppression("Relay log write failure: could not queue event from master");
+
+--echo #
+--echo # Initialize test data
+--echo #
+--connection master
+create table t1 (a int);
+insert into t1 values (1);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+# Both cases need a heartbeat to arrive promptly. The default period is
+# slave_net_timeout/2. The CHANGE MASTER that rpl_end.inc runs on every
+# server restores it.
+--source include/stop_slave.inc
+CHANGE MASTER TO MASTER_HEARTBEAT_PERIOD= 2;
+--source include/start_slave.inc
+
+--connection master
+set @saved_dbug= @@global.debug_dbug;
+
+--echo #
+--echo # Test Case 1: a heartbeat that leaves no room for a file name
+--echo #
+
+# The dump thread reads the injection when the replica reconnects below.
+--connection master
+set @@global.debug_dbug= "+d,binlog_sender_truncated_heartbeat";
+
+--connection slave
+--source include/stop_slave.inc
+--source include/start_slave.inc
+
+--echo # Waiting for the IO thread to reject the truncated heartbeat
+# queue_event() reports 1623, then handle_slave_io() overwrites it with 1595
+--let $slave_io_errno= 1595
+--let $show_slave_io_error= 1
+--source include/wait_for_slave_io_error.inc
+
+--echo # Ensure heartbeat validation rejected the event
+--let $assert_text= The IO thread reported an inconsistent heartbeat event
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= Unexpected master.s heartbeat data: inconsistent heartbeat event content
+--let $assert_count= 1
+--source include/assert_grep.inc
+
+# An empty log_file_name shows ident_len was bounded. Reaching this assertion
+# at all is the other half of the evidence.
+--let $assert_text= The heartbeat error should report an empty log_file_name
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= heartbeat event content;the event.s data: log_file_name  log_pos [0-9]+
+--let $assert_count= 1
+--source include/assert_grep.inc
+
+--connection master
+set @@global.debug_dbug= @saved_dbug;
+
+--connection slave
+--source include/stop_slave_sql.inc
+--source include/start_slave.inc
+
+--echo #
+--echo # Test Case 2: a heartbeat shorter than the common header
+--echo #
+--connection master
+set @@global.debug_dbug= "+d,binlog_sender_undersized_heartbeat";
+
+--connection slave
+--source include/stop_slave.inc
+--source include/start_slave.inc
+
+--echo # Waiting for the IO thread to reject the undersized heartbeat
+--let $slave_io_errno= 1595
+--let $show_slave_io_error= 1
+--source include/wait_for_slave_io_error.inc
+
+--echo # Ensure the event was rejected before its header was read
+--let $assert_text= The IO thread reported an event shorter than the common header
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= Event from master is shorter than its common header; the event.s length: [0-9]+
+--let $assert_count= 1
+--source include/assert_grep.inc
+
+--connection master
+set @@global.debug_dbug= @saved_dbug;
+
+--connection slave
+--source include/stop_slave_sql.inc
+--source include/start_slave.inc
+
+--echo #
+--echo # Ensure replication works after the invalid heartbeat events
+--echo #
+--connection master
+insert into t1 values (2);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+--let $diff_tables=master:t1,slave:t1
+--source include/diff_tables.inc
+
+--echo #
+--echo # Cleanup
+--echo #
+--connection master
+drop table t1;
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+--source include/rpl_end.inc
+--echo # End of rpl_heartbeat_log_event_underflow.test
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 00a393a03955a..24237b3b03137 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -8818,20 +8818,34 @@ void Ignorable_log_event::pack_info(Protocol *protocol)
 #if defined(HAVE_REPLICATION)
 Heartbeat_log_event::Heartbeat_log_event(const uchar *buf, uint event_len,
                     const Format_description_log_event* description_event)
-  :Log_event(buf, description_event)
+  :Log_event(buf, description_event), ident_len(0), log_ident(NULL)
 {
-  uint8 header_size= description_event->common_header_len;
+  uint sub_header_len= (log_pos == 0) ? HB_SUB_HEADER_LEN : 0;
+  uint all_headers_len= description_event->common_header_len + sub_header_len;
+
+  /*
+    The comparison is <= rather than <, so an event whose length stops exactly
+    at the headers is rejected along with a shorter one.
+
+      * A shorter event must be rejected out of necessity. ident_len is
+        unsigned, so the subtraction below would wrap and the caller would
+        read the log file name from far past the event.
+      * An event of exactly the header length must be rejected as policy. It
+        carries no log file name, and a heartbeat that names no binary log
+        file tells the replica nothing it can compare its own coordinates
+        against.
+
+    Leaving log_ident at NULL is what reports both to the caller, through
+    is_valid().
+  */
+  if (event_len <= all_headers_len)
+    return;
+
   if (log_pos == 0)
-  {
-    log_pos= uint8korr(buf + header_size);
-    log_ident= buf + header_size + HB_SUB_HEADER_LEN;
-    ident_len= event_len - (header_size + HB_SUB_HEADER_LEN);
-  }
-  else
-  {
-    log_ident= buf + header_size;
-    ident_len = event_len - header_size;
-  }
+    log_pos= uint8korr(buf + description_event->common_header_len);
+
+  log_ident= buf + all_headers_len;
+  ident_len= event_len - all_headers_len;
 }
 #endif
 
diff --git a/sql/slave.cc b/sql/slave.cc
index 9ca95dee93851..d1deb78749a1f 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -6509,11 +6509,34 @@ static int queue_event(Master_info* mi, const uchar *buf, ulong event_len)
   const uchar *save_buf= NULL; // needed for checksumming the fake Rotate event
   uchar rot_buf[LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN + FN_REFLEN];
 
+  /* The shortest event a master of this binlog version can legally send. */
+  ulong min_event_len=
+    mi->rli.relay_log.description_event_for_queue->binlog_version < 4 ?
+    OLD_HEADER_LEN : LOG_EVENT_MINIMAL_HEADER_LEN;
+
   DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_OFF || 
               checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF || 
               checksum_alg == BINLOG_CHECKSUM_ALG_CRC32); 
 
   DBUG_ENTER("queue_event");
+
+  /*
+    Reject an event that is shorter than the common header before anything
+    reads that header. The code below indexes into the header unconditionally,
+    and several of the per-type parsers derive a body length by subtracting
+    the header length, which wraps on a shorter event.
+    event_checksum_test() subtracts BINLOG_CHECKSUM_LEN in the same way.
+  */
+  if (unlikely(event_len < min_event_len))
+  {
+    error= ER_SLAVE_FATAL_ERROR;
+    error_msg.append(STRING_WITH_LEN("Event from master is shorter than its "
+                                     "common header; the event's length: "));
+    error_msg.append_ulonglong(event_len);
+    unlock_data_lock= FALSE;
+    goto err;
+  }
+
   /*
     FD_queue checksum alg description does not apply in a case of
     FD itself. The one carries both parts of the checksum data.
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 25d5c9e0a585f..72626f162d8f3 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -863,10 +863,28 @@ static int send_heartbeat_event(binlog_send_info *info,
     event_len+= HB_SUB_HEADER_LEN;
   }
 
+  size_t hb_header_len= sizeof(header);
+#ifndef DBUG_OFF
+  /*
+    Corrupt the heartbeat event to test that a replica can correctly identify
+    malformed events. 17 bytes keeps LOG_POS_OFFSET on the wire and clears
+    queue_event()'s minimum length, so the heartbeat parser rejects it.
+    13 bytes falls under that minimum, so queue_event() rejects it first.
+  */
+  DBUG_EXECUTE_IF("binlog_sender_truncated_heartbeat", hb_header_len= 17;);
+  DBUG_EXECUTE_IF("binlog_sender_undersized_heartbeat", hb_header_len= 13;);
+  if (hb_header_len != sizeof(header))
+  {
+    ident_len= 0;
+    event_len= hb_header_len + (sub_header_in_use ? HB_SUB_HEADER_LEN : 0) +
+      (do_checksum ? BINLOG_CHECKSUM_LEN : 0);
+  }
+#endif
+
   int4store(header + EVENT_LEN_OFFSET, event_len);
   int2store(header + FLAGS_OFFSET, 0);
 
-  packet->append(header, sizeof(header));
+  packet->append(header, hb_header_len);
   if (sub_header_in_use)
     packet->append(sub_header_buf, sizeof(sub_header_buf));
   packet->append(p, ident_len);                    // log_file_name
@@ -874,7 +892,7 @@ static int send_heartbeat_event(binlog_send_info *info,
   if (do_checksum)
   {
     char b[BINLOG_CHECKSUM_LEN];
-    ha_checksum crc= my_checksum(0, (uchar*) header, sizeof(header));
+    ha_checksum crc= my_checksum(0, (uchar*) header, hb_header_len);
     if (sub_header_in_use)
       crc= my_checksum(crc, (uchar*) sub_header_buf, sizeof(sub_header_buf));
     crc= my_checksum(crc, (uchar*) p, ident_len);