Added tests for binlog_server.

mtr/binlog_streaming/include/diff_with_storage_object.inc

@@ -33,7 +33,7 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $downloaded_file_path = $MYSQL_TMP_DIR/diff_with_storage_object.downloaded
-  --exec $aws_cli s3 cp s3://$aws_s3_bucket$storage_object $downloaded_file_path > /dev/null
+  --exec $aws_cli s3 cp s3://$aws_s3_bucket/$storage_object $downloaded_file_path > /dev/null
   --diff_files $local_file $downloaded_file_path
   --remove_file $downloaded_file_path
 }

mtr/binlog_streaming/include/patch_binlog_in_use_flag.inc

@@ -0,0 +1,31 @@
+# Copy a binlog file and clear LOG_EVENT_BINLOG_IN_USE_F in the patched copy.
+#
+# Usage:
+#   --let $binlog_file_to_patch = $binlog_base_dir/$binlog_name
+#   --let PATCHED_BINLOG_FILE = $MYSQL_TMP_DIR/$binlog_name.patched
+#   --source ../include/patch_binlog_in_use_flag.inc
+
+--echo
+--echo *** Patching the server version of the second binlog file to clear the
+--echo *** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+--copy_file $binlog_file_to_patch $PATCHED_BINLOG_FILE
+
+--perl
+  use strict;
+  use warnings;
+  use constant MAGIC_OFFSET => 21;
+  my $binlog_file_perl = $ENV{'PATCHED_BINLOG_FILE'};
+
+  open(my $fh, '+<:raw', $binlog_file_perl) or die "Failed to open file: $!";
+
+  seek($fh, MAGIC_OFFSET, 0);
+  my $byte;
+  read($fh, $byte, 1);
+
+  $byte = ord($byte) & 0xFE;
+
+  seek($fh, MAGIC_OFFSET, 0);
+  print $fh pack('C', $byte);
+
+  close($fh);
+EOF

mtr/binlog_streaming/include/set_up_binsrv_environment.inc

@@ -38,11 +38,11 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $binsrv_buffer_path = $MYSQL_TMP_DIR/buffer
-  --let $binsrv_storage_path = `SELECT CONCAT('/mtr-', UUID())`
+  --let $binsrv_storage_path = `SELECT CONCAT('mtr-', UUID())`
   --let $aws_s3_bucket = $MTR_BINSRV_AWS_S3_BUCKET
   if ($MTR_BINSRV_AWS_S3_ENDPOINT != '')
   {
-    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/$MTR_BINSRV_AWS_S3_BUCKET', '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/$MTR_BINSRV_AWS_S3_BUCKET', '/$binsrv_storage_path');
   }
   if ($MTR_BINSRV_AWS_S3_ENDPOINT == '')
   {
@@ -51,7 +51,7 @@ if ($storage_backend == s3)
     {
       --let $qualified_bucket = $qualified_bucket.$MTR_BINSRV_AWS_S3_REGION
     }
-    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$qualified_bucket',  '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$qualified_bucket',  '/$binsrv_storage_path');
   }
 }
 

mtr/binlog_streaming/include/tear_down_binsrv_environment.inc

@@ -6,7 +6,7 @@ if ($storage_backend == file)
 }
 if ($storage_backend == s3)
 {
-  --exec $aws_cli s3 rm s3://$aws_s3_bucket$binsrv_storage_path/ --recursive > /dev/null
+  --exec $aws_cli s3 rm s3://$aws_s3_bucket/$binsrv_storage_path/ --recursive > /dev/null
   --force-rmdir $binsrv_buffer_path
 }
 

mtr/binlog_streaming/include/terminate_binsrv.inc

@@ -0,0 +1,19 @@
+# Terminate the Binlog Server Utility and wait until the process exits.
+#
+# Usage:
+#   --let $binsrv_pid = <pid>
+#   --source ../include/terminate_binsrv.inc
+
+--exec kill -s TERM $binsrv_pid
+--let EXPORTED_BINSRV_PID = $binsrv_pid
+--perl
+  use strict;
+  use warnings;
+  use Errno;
+  my $pid = $ENV{'EXPORTED_BINSRV_PID'};
+  my $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  while (!$not_present) {
+    sleep(1);
+    $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  }
+EOF

mtr/binlog_streaming/include/trigger_binsrv_pull.inc

@@ -0,0 +1,13 @@
+# Start the Binlog Server Utility pull command in background.
+#
+# Usage:
+#   --let $binsrv_spawn_cmd_line = $BINSRV pull ... & echo \$! > $binsrv_pid_file
+#   --source ../include/trigger_binsrv_pull.inc
+
+--let EXPORTED_BINSRV_SPAWN_CMD_LINE = $binsrv_spawn_cmd_line
+--perl
+  use strict;
+  use warnings;
+  my $cmd = $ENV{'EXPORTED_BINSRV_SPAWN_CMD_LINE'};
+  system("$cmd");
+EOF

mtr/binlog_streaming/r/binsrv.result

@@ -29,6 +29,10 @@ INSERT INTO t1 VALUES(DEFAULT);
 *** Checking that the Binlog Server utility detected an empty storage
 include/assert_grep.inc [Binlog storage must be initialized on an empty directory]
 
+*** Verifying that binlog files were written to storage
+include/assert_grep.inc [First binlog should exist in storage]
+include/assert_grep.inc [Second binlog also hould exist in storage]
+
 *** Comparing server and downloaded versions of the first binlog file.
 
 *** Patching the server version of the second binlog file to clear the

mtr/binlog_streaming/r/checkpointing_verification.result

@@ -0,0 +1,116 @@
+*** Testing checkpointing verification
+*** For S3 backend, each checkpoint triggers a full object re-upload.
+
+*** Resetting replication at the very beginning of the test.
+
+*** Creating a table and generating large binlog events
+CREATE TABLE t1(
+id SERIAL,
+val CHAR(64) CHARACTER SET ascii NOT NULL,
+PRIMARY KEY(id)
+) ENGINE=InnoDB;
+
+*** Generating data to trigger checkpointing
+
+*** Determining current open binlog file name
+
+*** Setting up checkpointing configuration
+
+*** Setting up configuration
+SET @save_binsrv_checkpoint_size = 'binsrv_checkpoint_size';
+SET @save_binsrv_checkpoint_interval = 'binsrv_checkpoint_interval';
+
+*** Test 1: Test checkpointing with existing open binary log
+*** Expectation:
+*** - Without checkpointing: storage object stays at magic payload size while binsrv runs
+*** - With size checkpointing: storage object grows beyond magic payload while binsrv runs
+*** Note: Checkpointing only works with S3 backend, not with file backend
+
+Executing binlog_server in pull mode without checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+Executing binlog_server in pull mode with checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+*** Test 2: Test that new updates in binary log on server will be flushed
+*** to the binlog_server storage directory due to checkpointing.
+
+
+*** Starting pull mode in background to monitor checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+
+*** Capturing current storage object size (before workload)
+include/read_file_to_var.inc
+
+*** Generating more data to trigger checkpointing during pull
+
+*** Waiting a bit to allow checkpointing to occur
+
+*** Capturing storage object size after checkpointing workload
+include/read_file_to_var.inc
+
+*** Verifying checkpointing occurred (by storage object size change)
+
+*** Patching the server version of the binlog file to clear the
+*** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+*** (The binlog is still open on the server, so we patch it instead of flushing
+*** to avoid adding a ROTATE event that wouldn't exist in the storage object)
+
+*** Terminating pull mode
+
+*** Verifying binlog file integrity after checkpointing
+
+*** Starting pull mode again to verify it resumes correctly without any issues.
+include/read_file_to_var.inc
+
+*** Adding more data
+INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+FLUSH BINARY LOGS;
+
+*** Waiting a bit then terminating
+
+*** Verifying that resume worked correctly (no duplicate data)
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;

mtr/binlog_streaming/r/pull_mode.result

@@ -60,4 +60,61 @@ FLUSH BINARY LOGS;
 *** Removing the Binlog Server utility log file.
 
 *** Removing the Binlog Server utility configuration file.
-KILL CONNECTION <CONNECTION_ID>;
+
+*** Pull mode: zero idle_time
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background in pull mode with zero idle time
+include/read_file_to_var.inc
+
+*** Creating data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+*** Waiting for read timeout to trigger reconnection.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+*** Pull mode: very short timeouts and rapid reconnection
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background
+include/read_file_to_var.inc
+
+*** Creating data and restarting server multiple times
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;

mtr/binlog_streaming/r/read_timeout_flush.result

@@ -0,0 +1,84 @@
+
+*** Test : Storage flush after read_timeout disconnection (PS-10320 fix)
+*** This test verifies that local binary log data is flushed to storage
+*** upon disconnection due to read_timeout
+*** This test also verifies existence of temporary files in buffer directory
+*** in S3 storage mode while binlog_server run is in progress.
+SET SESSION binlog_transaction_compression=ON;
+CREATE TABLE t2(id INT UNSIGNED NOT NULL AUTO_INCREMENT, val CHAR(64), PRIMARY KEY(id)) ENGINE=InnoDB;
+
+*** Starting with a fresh binlog
+FLUSH BINARY LOGS;
+
+*** Setting up binsrv with short read_timeout to trigger timeout
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background in pull mode
+include/read_file_to_var.inc
+
+*** Waiting for binsrv to initialize and create binlog file in storage
+
+*** Step 1: Check initial binlog size (baseline)
+include/read_file_to_var.inc
+
+*** Step 2: Execute INSERT and wait for read_timeout to trigger flush (read_timeout=20, wait 25 seconds)
+INSERT INTO t2(val) VALUES('test_data_before_timeout');
+
+*** Step 3: Check binlog size after read_timeout flush (should have increased)
+include/read_file_to_var.inc
+
+*** Verifying that data was flushed to storage after first read_timeout (PS-10320 fix)
+
+*** Step 4: Wait around idle_time and generate new data in the binary log. verify either in idle_time or upon reconnection data was not flushed to storage.
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+SET @save_binlog_row_value_options = @@GLOBAL.binlog_row_value_options;
+SET GLOBAL binlog_row_value_options = PARTIAL_JSON;
+Warnings:
+Warning	3647	When binlog_row_image=FULL, the option binlog_row_value_options=PARTIAL_JSON will be used only for the after-image. Full values will be written in the before-image, so the saving in disk space due to binlog_row_value_options is limited to less than 50%.
+UPDATE t2 set val='update_test_data_after_timeout' where val='test_data_after_timeout';
+CREATE TABLE tj (j JSON);
+INSERT INTO tj VALUES('{"name":"john","age":30,"city":"blr"}');
+INSERT INTO tj VALUES('{"arr":[1,2,3],"details":{"salary":5000}}');
+UPDATE tj SET j = JSON_REPLACE(j, '$.city', 'mysore') WHERE JSON_EXTRACT(j, '$.name') = 'john';
+UPDATE tj SET j = JSON_REPLACE(j, '$.arr[1]', 999);
+DROP TABLE tj;
+SET GLOBAL binlog_row_value_options = @save_binlog_row_value_options;
+
+*** Step 5: Check binlog size after around idle time (should NOT have changed - since no flush)
+include/read_file_to_var.inc
+
+*** Verifying that size did NOT change after reconnection (no flush)
+
+*** Step 6: Execute INSERT again
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+
+*** Step 7: Check size immediately after INSERT ie around idle/reconnection time (should NOT have changed - data in buffer)
+include/read_file_to_var.inc
+
+*** Verifying that size did NOT change immediately after INSERT (data still in buffer)
+
+*** Step 8: Wait for read_timeout to trigger flush again (read_timeout=20)
+
+*** Step 9: Check binlog size after second read_timeout flush (should have increased)
+include/read_file_to_var.inc
+
+*** Verifying that size increased after second read_timeout flush
+FLUSH BINARY LOGS;
+
+*** Terminating the utility
+
+*** Final verification: Compare storage object with server binlog
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t2;