MDEV-27939 Log buffer wrap-around errors on PMEM

When the log is stored in persistent memory, log_sys.buf[] is
a ring buffer that directly maps to the circular ib_logfile0 file.

There were several errors that could occur in the special case
when a log record ends exactly at the end of the log file and the
next record would start at log_sys.buf[log_sys.START_OFFSET].

mariabackup.huge_lsn,strict_full_crc32: Write the first record
at the very end of the circular file, to reproduce the failure
scenarios.

recv_sys_t::parse(): On PMEM, wrap the end offset of the record
from log_sys.file_size to log_sys.START_OFFSET if needed.
Otherwise, both InnoDB recovery and mariadb-backup would try
to parse the next record from an invalid address.

filename_to_spacename(): Remove an assumption about the format
of file names. While the server currently writes file names like
./databasename/tablename.ibd we might want to stop writing the
redundant ./ prefix in the future. The test mariabackup.huge_lsn
is generating such file names.

xtrabackup_copy_logfile(): Correctly copy a record that ends at
the very end of the log_sys.buf[].

The errors in mariadb-backup were reproduced with the test
mariabackup.huge_lsn,strict_full_crc32 and an additional patch
to use the start checkpoint of the test:

diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 27dce5fa17d..e17a1692d6f 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -1796,7 +1796,8 @@ dberr_t recv_sys_t::find_checkpoint()
         continue;
       }

-      if (checkpoint_lsn >= log_sys.next_checkpoint_lsn)
+      if (checkpoint_lsn >= log_sys.next_checkpoint_lsn &&
+          checkpoint_lsn != 0x1000fffffe10)
       {
         log_sys.next_checkpoint_lsn= checkpoint_lsn;
         log_sys.next_checkpoint_no= field == log_t::CHECKPOINT_1;

extra/mariabackup/xtrabackup.cc

@@ -803,9 +803,8 @@ static std::string filename_to_spacename(const void *filename, size_t len)
 	ut_a(table);
 	*table = 0;
 	char *db = strrchr(f, '/');
-	ut_a(db);
 	*table = '/';
-	std::string s(db+1);
+	std::string s(db ? db+1 : f);
 	free(f);
 	return s;
 }
@@ -3032,8 +3031,8 @@ static bool xtrabackup_copy_logfile()
                 ds_write(dst_log_file, &seq_1, 1))
               goto write_error;
             if (so < -1 &&
-                ds_write(dst_log_file, log_sys.buf + recv_sys.len - (1 - so),
-                         1 - so))
+                ds_write(dst_log_file, log_sys.buf + recv_sys.len + (1 + so),
+                         -(1 + so)))
               goto write_error;
             if (ds_write(dst_log_file, log_sys.buf + log_sys.START_OFFSET,
                          recv_sys.offset - log_sys.START_OFFSET))

mysql-test/suite/mariabackup/huge_lsn,strict_crc32.rdiff

@@ -1,8 +1,10 @@
-@@ -2,7 +2,7 @@
+@@ -1,8 +1,8 @@
+ #
  # MDEV-13416 mariabackup fails with EFAULT "Bad Address"
  #
- # restart
--FOUND 1 /redo log from 2\.012MiB to [0-9.]*[KMGT]iB; LSN=17596481011216\b/ in mysqld.1.err
+-# restart: --innodb-log-file-size=4M --innodb-encrypt-log=0
+-FOUND 1 /InnoDB: log sequence number 17596481011216/ in mysqld.1.err
++# restart
 +FOUND 1 /redo log: [0-9.]*[KMGT]iB; LSN=17596481010687\b/ in mysqld.1.err
  CREATE TABLE t(i INT) ENGINE INNODB;
  INSERT INTO t VALUES(1);

mysql-test/suite/mariabackup/huge_lsn.result

@@ -1,8 +1,8 @@
 #
 # MDEV-13416 mariabackup fails with EFAULT "Bad Address"
 #
-# restart
-FOUND 1 /redo log from 2\.012MiB to [0-9.]*[KMGT]iB; LSN=17596481011216\b/ in mysqld.1.err
+# restart: --innodb-log-file-size=4M --innodb-encrypt-log=0
+FOUND 1 /InnoDB: log sequence number 17596481011216/ in mysqld.1.err
 CREATE TABLE t(i INT) ENGINE INNODB;
 INSERT INTO t VALUES(1);
 # xtrabackup backup

mysql-test/suite/mariabackup/huge_lsn.test

@@ -40,27 +40,33 @@ do "$ENV{MTR_SUITE_DIR}/../innodb/include/crc32.pl";
 my $file= "$ENV{MYSQLD_DATADIR}/ib_logfile0";
 open(FILE, ">", $file) or die "Unable to open $file\n";
 binmode FILE;
-# the desired log sequence number, plus 16
 my $extra_repeat = 139820;
+# The desired log sequence number is 17596481011216 (0x1000fffffe10).
+my $file_size=4<<20;
 my $lsn_hi=4096,$lsn_lo=0xfffffe00 - $extra_repeat * 15;
 my $polynomial = 0x82f63b78; # CRC-32C
 my ($header, $checkpoint, $log);
-$header = "Phys" . pack("x[4]NN", $lsn_hi, $lsn_lo) .
+$header = "Phys" . pack("x[4]NN", $lsn_hi, $lsn_lo - $file_size + 0x300f) .
           "some Perl code" . pack("x[478]");
 $header .= pack("Nx[3584]", mycrc32($header, 0, $polynomial));
 $checkpoint = pack("NNNNx[44]", $lsn_hi, $lsn_lo, $lsn_hi, $lsn_lo);
 $checkpoint .= pack("Nx[8128]", mycrc32($checkpoint, 0, $polynomial));
 $log = pack("CxxNN", 0xfa, $lsn_hi, $lsn_lo);
-$log .= pack("CN", 1, mycrc32($log, 0, $polynomial));
+$log .= pack("CN", 0, mycrc32($log, 0, $polynomial));
 
 # Write more than 2MiB of FILE_MODIFY mini-transactions to exercise the parser.
 my $extra = pack("CCxa*", 0xb9, 127, "a/b.ibd");
-$extra .= pack("CN", 1, mycrc32($extra, 0, $polynomial));
+$extra .= pack("CN", 0, mycrc32($extra, 0, $polynomial));
 
-print FILE $header, $checkpoint, $extra x $extra_repeat, $log;
+print FILE $header, $checkpoint, $extra x ($extra_repeat - 1), $log;
+seek(FILE, $file_size - 15, 0);
+$extra = pack("CCxa*", 0xb9, 127, "c/d.ibd");
+$extra .= pack("CN", 1, mycrc32($extra, 0, $polynomial));
+print FILE $extra;
 close(FILE) or die "Unable to close $file\n";
 EOF
---let SEARCH_PATTERN= redo log from 2\\.012MiB to [0-9.]*[KMGT]iB; LSN=17596481011216\\b
+--let SEARCH_PATTERN= InnoDB: log sequence number 17596481011216
+--let $restart_parameters=--innodb-log-file-size=4M --innodb-encrypt-log=0
 }
 
 --source include/start_mysqld.inc
@@ -80,6 +86,7 @@ INSERT INTO t VALUES(2);
 echo # xtrabackup prepare;
 --disable_result_log
 exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+--let $restart_parameters=
 --source include/restart_and_restore.inc
 --enable_result_log
 SELECT * FROM t;

storage/innobase/log/log0recv.cc

@@ -2009,6 +2009,8 @@ static void store_freed_or_init_rec(page_id_t page_id, bool freed)
 /** Wrapper for log_sys.buf[] between recv_sys.offset and recv_sys.len */
 struct recv_buf
 {
+  bool is_pmem() const noexcept { return log_sys.is_pmem(); }
+
   const byte *ptr;
 
   constexpr recv_buf(const byte *ptr) : ptr(ptr) {}
@@ -2102,6 +2104,8 @@ struct recv_buf
 /** Ring buffer wrapper for log_sys.buf[]; recv_sys.len == log_sys.file_size */
 struct recv_ring : public recv_buf
 {
+  static constexpr bool is_pmem() { return true; }
+
   constexpr recv_ring(const byte *ptr) : recv_buf(ptr) {}
 
   constexpr static bool is_eof() { return false; }
@@ -2323,6 +2327,11 @@ inline recv_sys_t::parse_mtr_result recv_sys_t::parse(store_t store, source &l)
   ut_d(const source el{l});
   lsn+= l - begin;
   offset= l.ptr - log_sys.buf;
+  if (!l.is_pmem());
+  else if (offset == log_sys.file_size)
+    offset= log_sys.START_OFFSET;
+  else
+    ut_ad(offset < log_sys.file_size);
 
   ut_d(std::set<page_id_t> freed);
 #if 0 && defined UNIV_DEBUG /* MDEV-21727 FIXME: enable this */