From 24770088d1cc92d3c34039866f932c1c25f42ff4 Mon Sep 17 00:00:00 2001 From: Yura Sorokin Date: Tue, 6 Oct 2026 16:44:22 +0200 Subject: [PATCH 1/2] PBS-42 feature: Implement basic support for COM_BINLOG_DUMP packet handling (part 5) https://perconadev.atlassian.net/browse/PBS-42 Implemented "resume from the specified position" logic for position- based replication in the listening mode. Network service extended with handling SET @source_binlog_checksum = 'CRC32', @master_binlog_checksum = 'CRC32'; and SET @source_binlog_checksum = 'NONE', @master_binlog_checksum = 'NONE' statements so that after switching to replication we would know whether the very first artificial ROTATE event generated by us needs to include checksum or not. By default this information should be taken from the last seen / sent FORMAT_DESCRIPTION event, but the very first artificial ROTATE event is a special case. Reworked 'operations::sender_context' class - it now operates as a state machine. In addition, instead of generating absolutely context unaware artificial ROTATE and FORMAT_DESCRIPTION events, we now always fetch them from the real data files and perform only necessary field changes before sending them back to the replication channel. Reworked event generation helpers: they now have a lower level generate_xxx_event_ex() counterpart functions that do not require an instance of a 'binsrv::events::reader_context' class. Added new 'clear_element()' method to the 'util::flag_set<>' class template. 'binlog_streaming.rs_roundtrip' and 'binlog_streaming.rs_roundtrip_encryption' MTR test cases extended with one more 'mysqlbinlog' run that extracts new set of events from the specified binlog file position after they were generated on the server. In addition, not only do we try to restore the dump files generated by the 'mysqlbinlog' and check that the restored data is identical to the one generated directly, we now also perform event-to-event comparison between dumps generated from the PBS and directly from the MySQL Server. These test cases are also run with 4 combinations now: * binlog event checksums enabled / disabled on the server; * server ID left to default / set to custom. Fixed some typos in the code comments. --- README.md | 2 +- .../include/rs_roundtrip_body.inc | 116 +++- mtr/binlog_streaming/r/rs_roundtrip.result | 51 +- .../r/rs_roundtrip_encryption.result | 51 +- .../t/auth_method_switch.test | 4 +- .../t/rs_roundtrip.combinations | 13 + mtr/binlog_streaming/t/rs_roundtrip.test | 1 + .../t/rs_roundtrip_encryption.combinations | 13 + src/binsrv/basic_storage_backend.hpp | 2 +- src/binsrv/events/common_header.cpp | 12 +- .../format_description_post_header_impl.cpp | 11 + .../format_description_post_header_impl.hpp | 7 + src/binsrv/storage_core.cpp | 2 +- src/minimysql/network_service.cpp | 43 +- src/operations/collector_context.cpp | 12 +- src/operations/event_generation_helpers.cpp | 78 ++- src/operations/event_generation_helpers.hpp | 25 +- src/operations/sender_context.cpp | 502 +++++++++++++++--- src/operations/sender_context.hpp | 38 +- src/util/common_optional_types.hpp | 2 + src/util/flag_set.hpp | 5 + 21 files changed, 840 insertions(+), 150 deletions(-) create mode 100644 mtr/binlog_streaming/t/rs_roundtrip.combinations create mode 100644 mtr/binlog_streaming/t/rs_roundtrip_encryption.combinations diff --git a/README.md b/README.md index d2b57f91..35005920 100644 --- a/README.md +++ b/README.md @@ -712,7 +712,7 @@ Running the utility for the second time (in any mode) results in resuming stream ### Graceful termination -The user can request the utility operating in either `fetch` or `pull` mode to be gracefully terminated leaving storage in consistent state. For this, the utility sets custom handlers for the the following POSIX signals. +The user can request the utility operating in either `fetch` or `pull` mode to be gracefully terminated leaving storage in consistent state. For this, the utility sets custom handlers for the following POSIX signals. - `SIGINT` - for processing `^C` in console. - `SIGTERM` - for processing `kill `. diff --git a/mtr/binlog_streaming/include/rs_roundtrip_body.inc b/mtr/binlog_streaming/include/rs_roundtrip_body.inc index 4af21396..de1d80d6 100644 --- a/mtr/binlog_streaming/include/rs_roundtrip_body.inc +++ b/mtr/binlog_streaming/include/rs_roundtrip_body.inc @@ -14,10 +14,6 @@ # $binsrv_replication_source_port # $stmt_show_binary_log_status ---echo ---echo *** Determining the first binary log name. ---let $first_binlog = query_get_value($stmt_show_binary_log_status, File, 1) - --echo --echo *** Creating source_db.tbl and inserting the first batch of rows. CREATE DATABASE source_db; @@ -29,7 +25,7 @@ CREATE TABLE source_db.tbl( INSERT INTO source_db.tbl VALUES (1, 'one'), (2, 'two'), (3, 'three'); --echo ---echo *** Flushing the source binary log. +--echo *** Flushing the source binary log for the first time. FLUSH BINARY LOGS; --echo @@ -37,28 +33,107 @@ FLUSH BINARY LOGS; INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); --echo ---echo *** Fetching the source binary logs in position mode. +--echo *** Determining checkpoint binary log name and position. +--let $checkpoint_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1) +--let $checkpoint_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1) + +--echo +--echo *** Fetching the source binary logs in position mode for the first time. --exec $BINSRV fetch $binsrv_config_file_path > /dev/null --echo ---echo *** Starting Binlog Server Utility in background in pull mode. +--echo *** Starting Binlog Server Utility in background in pull mode for the first time. +--let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null +--source ../include/start_proc_in_background.inc + +--echo +--echo *** Waiting for the replication source listener port ready to accept +--echo *** connections for the first time. +--let $listening_port = $binsrv_replication_source_port +--source ../include/wait_for_listening_port.inc + +--echo +--echo *** Dumping the first part of binary logs from the Binlog Server +--echo *** replication source. +--let $first_roundtrip_dump = $MYSQL_TMP_DIR/first_rs_roundtrip.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port + --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password + --to-last-log "" > $first_roundtrip_dump 2>/dev/null; +--let $first_roundtrip_dump_filtered = $first_roundtrip_dump.filtered +--exec cat $first_roundtrip_dump | grep -E "^#[[:digit:]]+" > $first_roundtrip_dump_filtered + +--echo +--echo *** Dumping the first part of binary logs from the MySQL Server +--let $first_direct_dump = $MYSQL_TMP_DIR/first_rs_direct.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user + --to-last-log "" > $first_direct_dump 2>/dev/null; +--let $first_direct_dump_filtered = $first_direct_dump.filtered +--exec cat $first_direct_dump | grep -E "^#[[:digit:]]+" > $first_direct_dump_filtered + +--echo +--echo *** Comparing the first parts of binary logs from the PBS and MySQL Server +--diff_files $first_roundtrip_dump_filtered $first_direct_dump_filtered + +--echo +--echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for +--echo *** the process to terminate for the first time. +--source ../include/terminate_proc.inc + +--echo +--echo *** Inserting the third batch of rows. +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); + +--echo +--echo *** Flushing the source binary log for the second time. +FLUSH BINARY LOGS; + +--echo +--echo *** Inserting the fourth batch of rows. +INSERT INTO source_db.tbl VALUES (10, 'ten'), (11, 'eleven'), (12, 'twelve'); + +--echo +--echo *** Fetching the source binary logs in position mode for the second time. +--exec $BINSRV fetch $binsrv_config_file_path > /dev/null + +--echo *** Starting Binlog Server Utility in background in pull mode for the second time. --let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null --source ../include/start_proc_in_background.inc --echo --echo *** Waiting for the replication source listener port ready to accept ---echo *** connections. +--echo *** connections for the second time. --let $listening_port = $binsrv_replication_source_port --source ../include/wait_for_listening_port.inc --echo ---echo *** Dumping the binary logs from the Binlog Server replication source. ---let $roundtrip_dump = $MYSQL_TMP_DIR/rs_roundtrip.sql ---exec $MYSQL_BINLOG --read-from-remote-server --host=127.0.0.1 --port=$binsrv_replication_source_port --user=rpl --password=password --default-auth=caching_sha2_password $first_binlog > $roundtrip_dump 2>/dev/null +--echo *** Dumping the second part of binary logs from the Binlog Server +--echo *** replication source. +--let $second_roundtrip_dump = $MYSQL_TMP_DIR/second_rs_roundtrip.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port + --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password + --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $second_roundtrip_dump 2>/dev/null; +--let $second_roundtrip_dump_filtered = $second_roundtrip_dump.filtered +--exec cat $second_roundtrip_dump | grep -E "^#[[:digit:]]+" > $second_roundtrip_dump_filtered + +--echo +--echo *** Dumping the second part of binary logs from the MySQL Server +--let $second_direct_dump = $MYSQL_TMP_DIR/second_rs_direct.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user + --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $second_direct_dump 2>/dev/null; +--let $second_direct_dump_filtered = $second_direct_dump.filtered +--exec cat $second_direct_dump | grep -E "^#[[:digit:]]+" > $second_direct_dump_filtered + +--echo +--echo *** Comparing the second parts of binary logs from the PBS and MySQL Server +--diff_files $second_roundtrip_dump_filtered $second_direct_dump_filtered --echo --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for ---echo *** the process to terminate +--echo *** the process to terminate for the second time. --source ../include/terminate_proc.inc --echo @@ -68,8 +143,9 @@ RENAME TABLE source_db.tbl TO backup_db.tbl; DROP DATABASE source_db; --echo ---echo *** Restoring source_db from the replication source dump. ---exec $MYSQL < $roundtrip_dump +--echo *** Restoring source_db from two replication source dumps. +--exec $MYSQL < $first_roundtrip_dump +--exec $MYSQL < $second_roundtrip_dump --echo --echo *** Comparing the restored and backup tables. @@ -81,4 +157,14 @@ DROP DATABASE source_db; DROP DATABASE source_db; DROP DATABASE backup_db; ---remove_file $roundtrip_dump +--echo +--echo *** Cleaning up. +--remove_file $first_roundtrip_dump +--remove_file $first_roundtrip_dump_filtered +--remove_file $first_direct_dump +--remove_file $first_direct_dump_filtered + +--remove_file $second_roundtrip_dump +--remove_file $second_roundtrip_dump_filtered +--remove_file $second_direct_dump +--remove_file $second_direct_dump_filtered diff --git a/mtr/binlog_streaming/r/rs_roundtrip.result b/mtr/binlog_streaming/r/rs_roundtrip.result index 7cd43b27..574fa97b 100644 --- a/mtr/binlog_streaming/r/rs_roundtrip.result +++ b/mtr/binlog_streaming/r/rs_roundtrip.result @@ -8,8 +8,6 @@ *** Creating a temporary directory for storing *** binlog files downloaded via the Binlog Server utility. -*** Determining the first binary log name. - *** Creating source_db.tbl and inserting the first batch of rows. CREATE DATABASE source_db; CREATE TABLE source_db.tbl( @@ -19,31 +17,64 @@ PRIMARY KEY(id) ) ENGINE=InnoDB; INSERT INTO source_db.tbl VALUES (1, 'one'), (2, 'two'), (3, 'three'); -*** Flushing the source binary log. +*** Flushing the source binary log for the first time. FLUSH BINARY LOGS; *** Inserting the second batch of rows. INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); -*** Fetching the source binary logs in position mode. +*** Determining checkpoint binary log name and position. + +*** Fetching the source binary logs in position mode for the first time. + +*** Starting Binlog Server Utility in background in pull mode for the first time. +include/read_file_to_var.inc + +*** Waiting for the replication source listener port ready to accept +*** connections for the first time. + +*** Dumping the first part of binary logs from the Binlog Server +*** replication source. + +*** Dumping the first part of binary logs from the MySQL Server + +*** Comparing the first parts of binary logs from the PBS and MySQL Server -*** Starting Binlog Server Utility in background in pull mode. +*** Sending SIGTERM signal to the Binlog Server Utility and waiting for +*** the process to terminate for the first time. + +*** Inserting the third batch of rows. +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); + +*** Flushing the source binary log for the second time. +FLUSH BINARY LOGS; + +*** Inserting the fourth batch of rows. +INSERT INTO source_db.tbl VALUES (10, 'ten'), (11, 'eleven'), (12, 'twelve'); + +*** Fetching the source binary logs in position mode for the second time. +*** Starting Binlog Server Utility in background in pull mode for the second time. include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept -*** connections. +*** connections for the second time. + +*** Dumping the second part of binary logs from the Binlog Server +*** replication source. -*** Dumping the binary logs from the Binlog Server replication source. +*** Dumping the second part of binary logs from the MySQL Server + +*** Comparing the second parts of binary logs from the PBS and MySQL Server *** Sending SIGTERM signal to the Binlog Server Utility and waiting for -*** the process to terminate +*** the process to terminate for the second time. *** Moving source_db.tbl to backup_db.tbl. CREATE DATABASE backup_db; RENAME TABLE source_db.tbl TO backup_db.tbl; DROP DATABASE source_db; -*** Restoring source_db from the replication source dump. +*** Restoring source_db from two replication source dumps. *** Comparing the restored and backup tables. include/diff_tables.inc [source_db.tbl, backup_db.tbl] @@ -52,6 +83,8 @@ include/diff_tables.inc [source_db.tbl, backup_db.tbl] DROP DATABASE source_db; DROP DATABASE backup_db; +*** Cleaning up. + *** Removing the Binlog Server utility storage directory. *** Removing the Binlog Server utility log file. diff --git a/mtr/binlog_streaming/r/rs_roundtrip_encryption.result b/mtr/binlog_streaming/r/rs_roundtrip_encryption.result index e12630b6..74b7fe2c 100644 --- a/mtr/binlog_streaming/r/rs_roundtrip_encryption.result +++ b/mtr/binlog_streaming/r/rs_roundtrip_encryption.result @@ -10,8 +10,6 @@ *** Creating a temporary directory for storing *** binlog files downloaded via the Binlog Server utility. -*** Determining the first binary log name. - *** Creating source_db.tbl and inserting the first batch of rows. CREATE DATABASE source_db; CREATE TABLE source_db.tbl( @@ -21,31 +19,64 @@ PRIMARY KEY(id) ) ENGINE=InnoDB; INSERT INTO source_db.tbl VALUES (1, 'one'), (2, 'two'), (3, 'three'); -*** Flushing the source binary log. +*** Flushing the source binary log for the first time. FLUSH BINARY LOGS; *** Inserting the second batch of rows. INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); -*** Fetching the source binary logs in position mode. +*** Determining checkpoint binary log name and position. + +*** Fetching the source binary logs in position mode for the first time. + +*** Starting Binlog Server Utility in background in pull mode for the first time. +include/read_file_to_var.inc + +*** Waiting for the replication source listener port ready to accept +*** connections for the first time. + +*** Dumping the first part of binary logs from the Binlog Server +*** replication source. + +*** Dumping the first part of binary logs from the MySQL Server + +*** Comparing the first parts of binary logs from the PBS and MySQL Server -*** Starting Binlog Server Utility in background in pull mode. +*** Sending SIGTERM signal to the Binlog Server Utility and waiting for +*** the process to terminate for the first time. + +*** Inserting the third batch of rows. +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); + +*** Flushing the source binary log for the second time. +FLUSH BINARY LOGS; + +*** Inserting the fourth batch of rows. +INSERT INTO source_db.tbl VALUES (10, 'ten'), (11, 'eleven'), (12, 'twelve'); + +*** Fetching the source binary logs in position mode for the second time. +*** Starting Binlog Server Utility in background in pull mode for the second time. include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept -*** connections. +*** connections for the second time. + +*** Dumping the second part of binary logs from the Binlog Server +*** replication source. -*** Dumping the binary logs from the Binlog Server replication source. +*** Dumping the second part of binary logs from the MySQL Server + +*** Comparing the second parts of binary logs from the PBS and MySQL Server *** Sending SIGTERM signal to the Binlog Server Utility and waiting for -*** the process to terminate +*** the process to terminate for the second time. *** Moving source_db.tbl to backup_db.tbl. CREATE DATABASE backup_db; RENAME TABLE source_db.tbl TO backup_db.tbl; DROP DATABASE source_db; -*** Restoring source_db from the replication source dump. +*** Restoring source_db from two replication source dumps. *** Comparing the restored and backup tables. include/diff_tables.inc [source_db.tbl, backup_db.tbl] @@ -54,6 +85,8 @@ include/diff_tables.inc [source_db.tbl, backup_db.tbl] DROP DATABASE source_db; DROP DATABASE backup_db; +*** Cleaning up. + *** Removing the Binlog Server utility storage directory. *** Removing the Binlog Server utility log file. diff --git a/mtr/binlog_streaming/t/auth_method_switch.test b/mtr/binlog_streaming/t/auth_method_switch.test index 3a971806..94bb0f60 100644 --- a/mtr/binlog_streaming/t/auth_method_switch.test +++ b/mtr/binlog_streaming/t/auth_method_switch.test @@ -50,7 +50,7 @@ eval $stmt_reset_binary_logs_and_gtids; --echo *** (caching_sha2_password). No AuthMethodSwitch is expected on the --echo *** wire; a zero exit code from mysql confirms the session got as --echo *** far as running the probe query. ---exec $MYSQL --protocol=TCP --host=127.0.0.1 --port=$binsrv_replication_source_port --user=rpl --password=password --default-auth=caching_sha2_password --skip-column-names -e "$probe_query" >/dev/null 2>&1 +--exec $MYSQL --host=127.0.0.1 --port=$binsrv_replication_source_port --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password -e "$probe_query" >/dev/null 2>&1 --echo --echo *** Trigger: client forces mysql_native_password in its handshake so @@ -61,7 +61,7 @@ eval $stmt_reset_binary_logs_and_gtids; --echo *** MTR fails the --exec. The log grep that follows the shutdown --echo *** is what actually proves the switch happened - this line only --echo *** proves the session survived it. ---exec $MYSQL --protocol=TCP --host=127.0.0.1 --port=$binsrv_replication_source_port --user=rpl --password=password --default-auth=mysql_native_password --skip-column-names -e "$probe_query" >/dev/null 2>&1 +--exec $MYSQL --host=127.0.0.1 --port=$binsrv_replication_source_port --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=mysql_native_password -e "$probe_query" >/dev/null 2>&1 --echo --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for diff --git a/mtr/binlog_streaming/t/rs_roundtrip.combinations b/mtr/binlog_streaming/t/rs_roundtrip.combinations new file mode 100644 index 00000000..6d42b428 --- /dev/null +++ b/mtr/binlog_streaming/t/rs_roundtrip.combinations @@ -0,0 +1,13 @@ +[srv_nocrc_default_id] +binlog-checksum = NONE + +[srv_nocrc_custom_id] +binlog-checksum = NONE +server-id = 67 + +[srv_crc_default_id] +binlog-checksum = CRC32 + +[srv_crc_custom_id] +binlog-checksum = CRC32 +server-id = 67 diff --git a/mtr/binlog_streaming/t/rs_roundtrip.test b/mtr/binlog_streaming/t/rs_roundtrip.test index da60191a..75daa713 100644 --- a/mtr/binlog_streaming/t/rs_roundtrip.test +++ b/mtr/binlog_streaming/t/rs_roundtrip.test @@ -16,6 +16,7 @@ eval $stmt_reset_binary_logs_and_gtids; --let $binsrv_idle_time = 1 --let $binsrv_verify_checksum = TRUE --let $binsrv_replication_mode = position +--let $binsrv_checkpoint_size = 1 --source ../include/set_up_binsrv_environment.inc --source ../include/rs_roundtrip_body.inc diff --git a/mtr/binlog_streaming/t/rs_roundtrip_encryption.combinations b/mtr/binlog_streaming/t/rs_roundtrip_encryption.combinations new file mode 100644 index 00000000..6d42b428 --- /dev/null +++ b/mtr/binlog_streaming/t/rs_roundtrip_encryption.combinations @@ -0,0 +1,13 @@ +[srv_nocrc_default_id] +binlog-checksum = NONE + +[srv_nocrc_custom_id] +binlog-checksum = NONE +server-id = 67 + +[srv_crc_default_id] +binlog-checksum = CRC32 + +[srv_crc_custom_id] +binlog-checksum = CRC32 +server-id = 67 diff --git a/src/binsrv/basic_storage_backend.hpp b/src/binsrv/basic_storage_backend.hpp index e40d66a2..041694bf 100644 --- a/src/binsrv/basic_storage_backend.hpp +++ b/src/binsrv/basic_storage_backend.hpp @@ -50,7 +50,7 @@ class basic_storage_backend { // reader either sees the previous bytes in full or the new bytes in // full, never a partial mix. void put_object(std::string_view name, util::const_byte_span content); - // 'resize_object' is supposed to set a new size for the the specified + // 'resize_object' is supposed to set a new size for the specified // object void resize_object(std::string_view name, std::uint64_t new_size); // Single-object remove followed by a durability barrier. On diff --git a/src/binsrv/events/common_header.cpp b/src/binsrv/events/common_header.cpp index 64aa9045..48db2977 100644 --- a/src/binsrv/events/common_header.cpp +++ b/src/binsrv/events/common_header.cpp @@ -69,12 +69,20 @@ common_header::common_header(util::const_byte_span portion) std::uint32_t offset, std::uint32_t event_size, const util::ctime_timestamp ×tamp, code_type type_code, std::uint32_t server_id, common_header_flag_set flags) noexcept { - // artificial ROTATE event must have next_event_position set to zero + // artificial ROTATE and FORMAT_DESCRIPTION events must have + // next_event_position set to zero const std::uint32_t next_event_position{ - type_code == code_type::rotate && + (type_code == code_type::rotate || + type_code == code_type::format_description) && flags.has_element(common_header_flag_type::artificial) ? 0U : offset + event_size}; + // 'artificial' flag for FORMAT_DESCRIPTION event is used only as a marker + // for this function to set 'next_event_position' to 0 - it should not be + // serialized as part of the flags + if (type_code == code_type::format_description) { + flags.clear_element(common_header_flag_type::artificial); + } return common_header{timestamp, type_code, server_id, event_size, next_event_position, flags}; } diff --git a/src/binsrv/events/format_description_post_header_impl.cpp b/src/binsrv/events/format_description_post_header_impl.cpp index 6089ec70..8f8552eb 100644 --- a/src/binsrv/events/format_description_post_header_impl.cpp +++ b/src/binsrv/events/format_description_post_header_impl.cpp @@ -180,6 +180,17 @@ generic_post_header_impl::get_create_timestamp() return get_create_timestamp().simple_str(); } +void generic_post_header_impl:: + set_create_timestamp( + const util::ctime_timestamp &create_timestamp) noexcept { + set_create_timestamp_raw( + static_cast(create_timestamp.get_value())); +} +void generic_post_header_impl:: + set_readable_create_timestamp(std::string_view create_timestamp) { + set_create_timestamp(util::ctime_timestamp{create_timestamp}); +} + void generic_post_header_impl::encode_to( util::byte_span &destination) const { if (std::size(destination) < calculate_encoded_size()) { diff --git a/src/binsrv/events/format_description_post_header_impl.hpp b/src/binsrv/events/format_description_post_header_impl.hpp index b509bfd9..b4d4848f 100644 --- a/src/binsrv/events/format_description_post_header_impl.hpp +++ b/src/binsrv/events/format_description_post_header_impl.hpp @@ -78,6 +78,13 @@ class [[nodiscard]] generic_post_header_impl { [[nodiscard]] util::ctime_timestamp get_create_timestamp() const noexcept; [[nodiscard]] std::string get_readable_create_timestamp() const; + void set_create_timestamp_raw(std::uint32_t create_timestamp) noexcept { + create_timestamp_ = create_timestamp; + } + void + set_create_timestamp(const util::ctime_timestamp &create_timestamp) noexcept; + void set_readable_create_timestamp(std::string_view create_timestamp); + [[nodiscard]] std::uint8_t get_common_header_length_raw() const noexcept { return common_header_length_; } diff --git a/src/binsrv/storage_core.cpp b/src/binsrv/storage_core.cpp index ad25b8fa..18161404 100644 --- a/src/binsrv/storage_core.cpp +++ b/src/binsrv/storage_core.cpp @@ -1006,7 +1006,7 @@ storage_core::generate_binlog_encryption_record() const { // ('' parameter) const auto &keyring_record{keyring_->get_key(active_kek_id_)}; - // identifying the the cipher name and the key data from the + // identifying the cipher name and the key data from the // keyring record - this data will be used to encrypt random file // keys generated for new binlog data files const auto &kek_cipher{keyring_record.get<"cipher">()}; diff --git a/src/minimysql/network_service.cpp b/src/minimysql/network_service.cpp index 7d1a6a47..dd34b807 100644 --- a/src/minimysql/network_service.cpp +++ b/src/minimysql/network_service.cpp @@ -288,16 +288,19 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) { // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters) const boost::asio::ip::tcp::endpoint &remote_endpoint, // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters) - const std::string &remote_endpoint_str, - std::chrono::seconds write_timeout) { + const std::string &remote_endpoint_str, std::chrono::seconds write_timeout, + bool session_source_binlog_checksum) { static constexpr auto block_size{1048576UZ}; static constexpr auto blocking_poll_interval{std::chrono::milliseconds{500}}; const bool blocking{!context.check_binlog_non_blocking_dump()}; - // TODO: initialize sender_context with binlog_name:position extracted - // from the COM_BINLOG_DUMP command. - operations::sender_context sender_ctx{logger, storage, block_size}; + const auto &binlog_name{context.get_binlog_filename()}; + const auto position{context.get_binlog_position()}; + + operations::sender_context sender_ctx{ + logger, storage, block_size, + binlog_name, position, session_source_binlog_checksum}; boost::asio::steady_timer idle_timer{ co_await boost::asio::this_coro::executor}; util::const_byte_span event_data{}; @@ -481,8 +484,18 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) { minimysql::connection_context & context)>; using query_container = std::unordered_map; - const auto set_checksum_query_handler = - [](minimysql::connection_context &ctx) { + bool session_source_binlog_checksum{false}; + + const auto set_checksum_on_query_handler = + [&session_source_binlog_checksum](minimysql::connection_context &ctx) { + session_source_binlog_checksum = true; + minimysql::network_buffer_container resultset; + resultset.emplace_back(ctx.generate_encoded_ok()); + return resultset; + }; + const auto set_checksum_off_query_handler = + [&session_source_binlog_checksum](minimysql::connection_context &ctx) { + session_source_binlog_checksum = false; minimysql::network_buffer_container resultset; resultset.emplace_back(ctx.generate_encoded_ok()); return resultset; @@ -525,10 +538,16 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) { }}, {"SET @source_binlog_checksum = 'NONE', @master_binlog_checksum = " "'NONE'", - set_checksum_query_handler}, + set_checksum_off_query_handler}, {"SET @master_binlog_checksum = 'NONE', @source_binlog_checksum = " "'NONE'", - set_checksum_query_handler}}; + set_checksum_off_query_handler}, + {"SET @source_binlog_checksum = 'CRC32', @master_binlog_checksum = " + "'CRC32'", + set_checksum_on_query_handler}, + {"SET @master_binlog_checksum = 'CRC32', @source_binlog_checksum = " + "'CRC32'", + set_checksum_on_query_handler}}; // starting command loop bool terminated{false}; @@ -576,9 +595,9 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) { std::size(ok_after_ping), remote_endpoint_str); } break; case minimysql::client_command_type::binlog_dump: { - co_await handle_binlog_dump_command(logger, storage, socket, context, - remote_endpoint, - remote_endpoint_str, write_timeout); + co_await handle_binlog_dump_command( + logger, storage, socket, context, remote_endpoint, + remote_endpoint_str, write_timeout, session_source_binlog_checksum); terminated = true; } break; case minimysql::client_command_type::quit: { diff --git a/src/operations/collector_context.cpp b/src/operations/collector_context.cpp index c1723a32..e06a587c 100644 --- a/src/operations/collector_context.cpp +++ b/src/operations/collector_context.cpp @@ -378,28 +378,30 @@ void collector_context::rewrite_and_process_binlog_event( offset = static_cast(storage_->get_current_position()); const auto generated_rotate_event_v{generate_rotate_event( event_buffer, context, offset, true /* current timestamp */, - server_id, false /* non-artificial */, binlog_name)}; + server_id, false /* non-artificial */, binlog_name, + binsrv::events::magic_binlog_offset)}; logger_->log(binsrv::log_severity::info, "rewrite: generated rotate event in the rewrite mode"); process_binlog_event(generated_rotate_event_v, context); } // generate and process ROTATE(artificial) event - offset = 0U; + offset = binsrv::events::magic_binlog_offset; // artificial ROTATE event must include zero timestamp const auto generated_artificial_rotate_event_v{generate_rotate_event( event_buffer, context, offset, false /* zero timestamp */, server_id, - true /* artificial */, binlog_name)}; + true /* artificial */, binlog_name, + binsrv::events::magic_binlog_offset)}; logger_->log( binsrv::log_severity::info, "rewrite: generated artificial rotate event in the rewrite mode"); process_binlog_event(generated_artificial_rotate_event_v, context); // generate and process FORMAT_DESCRIPTION event - offset = binsrv::events::magic_binlog_offset; const auto generated_format_description_event_v{ generate_format_description_event(event_buffer, context, offset, - server_id)}; + server_id, true /* enable checksum */, + false /* non-artificial */)}; logger_->log( binsrv::log_severity::info, "rewrite: generated format description event in the rewrite mode"); diff --git a/src/operations/event_generation_helpers.cpp b/src/operations/event_generation_helpers.cpp index 4c8e90d0..0129b488 100644 --- a/src/operations/event_generation_helpers.cpp +++ b/src/operations/event_generation_helpers.cpp @@ -33,13 +33,13 @@ namespace operations { -[[nodiscard]] binsrv::events::event_view generate_rotate_event( - binsrv::events::event_storage &event_buffer, - const binsrv::events::reader_context &context, std::uint32_t offset, - bool current_timestamp, std::uint32_t server_id, bool artificial, - const binsrv::events::composite_binlog_name &binlog_name) { +void generate_rotate_event_ex( + binsrv::events::event_storage &event_buffer, bool include_checksum, + std::uint32_t offset, bool current_timestamp, std::uint32_t server_id, + bool artificial, const binsrv::events::composite_binlog_name &binlog_name, + std::uint64_t position) { const binsrv::events::generic_post_header - post_header{binsrv::events::magic_binlog_offset}; + post_header{position}; const binsrv::events::generic_body body{ binlog_name}; @@ -53,46 +53,80 @@ namespace operations { flags |= binsrv::events::common_header_flag_type::artificial; } + const auto generated_event{ + binsrv::events::event::create_event( + offset, timestamp, server_id, flags, post_header, body, + include_checksum, event_buffer)}; +} + +[[nodiscard]] binsrv::events::event_view +generate_rotate_event(binsrv::events::event_storage &event_buffer, + const binsrv::events::reader_context &context, + std::uint32_t offset, bool current_timestamp, + std::uint32_t server_id, bool artificial, + const binsrv::events::composite_binlog_name &binlog_name, + std::uint64_t position) { + // the value of the 'include_checksum' parameters is taken from the // 'reader_context': immediately after reconnection it will be equal to // the '' configuration parameter and after // that will be taken from the FORMAT_DESCRIPTION events, which in the // rewrite mode will be generated by us and therefore will always include // 'checksum_algorithm' set to 'crc32' - const auto generated_event{ - binsrv::events::event::create_event( - offset, timestamp, server_id, flags, post_header, body, - context.is_footer_expected(), event_buffer)}; + generate_rotate_event_ex(event_buffer, context.is_footer_expected(), offset, + current_timestamp, server_id, artificial, + binlog_name, position); return binsrv::events::event_view{context, util::const_byte_span{event_buffer}}; } -[[nodiscard]] binsrv::events::event_view -generate_format_description_event(binsrv::events::event_storage &event_buffer, - const binsrv::events::reader_context &context, - std::uint32_t offset, - std::uint32_t server_id) { - const util::semantic_version server_version{ - context.get_current_encoded_server_version()}; +void generate_format_description_event_ex( + binsrv::events::event_storage &event_buffer, + const util::semantic_version &server_version, std::uint32_t offset, + std::uint32_t server_id, bool enable_checksum_algorithm, bool artificial) { const binsrv::events::generic_post_header< binsrv::events::code_type::format_description> post_header{ binsrv::events::default_binlog_version, server_version, - util::ctime_timestamp::now(), + artificial ? util::ctime_timestamp{} : util::ctime_timestamp::now(), binsrv::events::default_common_header_length, binsrv::events::reader_context::get_hardcoded_post_header_lengths( server_version.get_encoded())}; + + const auto checksum_algorithm{ + enable_checksum_algorithm ? binsrv::events::checksum_algorithm_type::crc32 + : binsrv::events::checksum_algorithm_type::off}; + const binsrv::events::generic_body< binsrv::events::code_type::format_description> - body{binsrv::events::checksum_algorithm_type::crc32}; + body{checksum_algorithm}; - // enforcing checksums for all rewritten upcoming events + // here we set the 'artificial' flag for the common header only as a marker + // for 'event::create_event<>()' - that function will make a decision + // whether to set 'next_event_position' to 0 based on this flag but will + // clear it out before serialization + binsrv::events::common_header_flag_set flags{}; + if (artificial) { + flags |= binsrv::events::common_header_flag_type::artificial; + } + + // FORMAT_DESCRIPTION events always include a checksum const auto generated_event{binsrv::events::event::create_event< binsrv::events::code_type::format_description>( - offset, util::ctime_timestamp::now(), server_id, - binsrv::events::common_header_flag_set{}, post_header, body, + offset, util::ctime_timestamp::now(), server_id, flags, post_header, body, true /* include_checksum */, event_buffer)}; +} +[[nodiscard]] binsrv::events::event_view generate_format_description_event( + binsrv::events::event_storage &event_buffer, + const binsrv::events::reader_context &context, std::uint32_t offset, + std::uint32_t server_id, bool enable_checksum_algorithm, bool artificial) { + const util::semantic_version server_version{ + context.get_current_encoded_server_version()}; + // enforcing checksums for all rewritten upcoming events + generate_format_description_event_ex(event_buffer, server_version, offset, + server_id, enable_checksum_algorithm, + artificial); return binsrv::events::event_view{context, util::const_byte_span{event_buffer}}; diff --git a/src/operations/event_generation_helpers.hpp b/src/operations/event_generation_helpers.hpp index d8e952be..72f61a68 100644 --- a/src/operations/event_generation_helpers.hpp +++ b/src/operations/event_generation_helpers.hpp @@ -25,20 +25,31 @@ #include "binsrv/gtids/gtid_set_fwd.hpp" +#include "util/semantic_version_fwd.hpp" + namespace operations { +void generate_rotate_event_ex( + binsrv::events::event_storage &event_buffer, bool include_checksum, + std::uint32_t offset, bool current_timestamp, std::uint32_t server_id, + bool artificial, const binsrv::events::composite_binlog_name &binlog_name, + std::uint64_t position); [[nodiscard]] binsrv::events::event_view generate_rotate_event(binsrv::events::event_storage &event_buffer, const binsrv::events::reader_context &context, std::uint32_t offset, bool current_timestamp, std::uint32_t server_id, bool artificial, - const binsrv::events::composite_binlog_name &binlog_name); - -[[nodiscard]] binsrv::events::event_view -generate_format_description_event(binsrv::events::event_storage &event_buffer, - const binsrv::events::reader_context &context, - std::uint32_t offset, - std::uint32_t server_id); + const binsrv::events::composite_binlog_name &binlog_name, + std::uint64_t position); + +void generate_format_description_event_ex( + binsrv::events::event_storage &event_buffer, + const util::semantic_version &server_version, std::uint32_t offset, + std::uint32_t server_id, bool enable_checksum_algorithm, bool artificial); +[[nodiscard]] binsrv::events::event_view generate_format_description_event( + binsrv::events::event_storage &event_buffer, + const binsrv::events::reader_context &context, std::uint32_t offset, + std::uint32_t server_id, bool enable_checksum_algorithm, bool artificial); [[nodiscard]] binsrv::events::event_view generate_previous_gtids_log_event(binsrv::events::event_storage &event_buffer, diff --git a/src/operations/sender_context.cpp b/src/operations/sender_context.cpp index c01c58fd..5e8b3e47 100644 --- a/src/operations/sender_context.cpp +++ b/src/operations/sender_context.cpp @@ -17,29 +17,59 @@ #include #include +#include #include #include +#include #include #include "binsrv/basic_logger.hpp" #include "binsrv/indexed_event_block.hpp" #include "binsrv/log_severity.hpp" +#include "binsrv/replication_mode_type.hpp" #include "binsrv/storage.hpp" +#include "binsrv/events/checksum_algorithm_type.hpp" +#include "binsrv/events/code_type.hpp" #include "binsrv/events/common_header_view.hpp" +#include "binsrv/events/composite_binlog_name.hpp" +#include "binsrv/events/event_view.hpp" +#include "binsrv/events/format_description_body_impl.hpp" +#include "binsrv/events/format_description_post_header_impl.hpp" +#include "binsrv/events/generic_body_fwd.hpp" +#include "binsrv/events/generic_post_header_fwd.hpp" #include "binsrv/events/protocol_traits_fwd.hpp" +#include "binsrv/events/reader_context.hpp" + +#include "operations/event_generation_helpers.hpp" #include "util/byte_span_fwd.hpp" +#include "util/common_optional_types.hpp" #include "util/dynamic_byte_buffer_fwd.hpp" namespace operations { sender_context::sender_context(binsrv::basic_logger_ptr logger, binsrv::storage_ptr storage, - std::size_t block_size) + std::size_t block_size, + std::string_view binlog_name, + std::uint64_t position, + bool session_source_binlog_checksum) : logger_{std::move(logger)}, storage_{std::move(storage)}, block_size_{block_size}, - range_{binsrv::events::magic_binlog_offset, block_size_} { + current_binlog_checksum_{session_source_binlog_checksum}, + fsm_state_{position == binsrv::events::magic_binlog_offset + ? fsm_state_type::start_from_beginning + : fsm_state_type::start_from_offset}, + position_for_artificial_rotate_{position}, + // an empty binlog name in the COM_BINLOG_DUMP request means "the + // oldest binlog file available" - in this case we keep 'binlog_name_' + // empty and let 'storage::fetch_event_block()' resolve it + binlog_name_{ + binlog_name.empty() + ? binsrv::events::composite_binlog_name{} + : binsrv::events::composite_binlog_name::parse(binlog_name)}, + range_{position, block_size_} { assert(block_size_ > 0UZ); assert(storage_); assert(logger_); @@ -48,81 +78,427 @@ sender_context::sender_context(binsrv::basic_logger_ptr logger, sender_context::~sender_context() = default; [[nodiscard]] bool sender_context::get_event(util::const_byte_span &event) { + // Additional (artificial) events that need to be generated. + + // When client requests replication from ["binlog.":4] or from ["":4] + // (from the magic offset, meaning the beginning of the binlog file), + // replication source must send the following sequence of events. + // 1. A generated artificial ROTATE with the 'position' field set to 4 and + // the 'binlog' field set to "binlog." (in case of an empty binlog + // name specified in the request, the 'binlog' field must be set to + // the oldest binlog file available on the server). Whether this message + // should include checksum or not must be determined from the + // '@source_binlog_checksum' / '@master_binlog_checksum' session + // variable set in MySQL connection before switching to replication + // mode. 'timestamp' and 'next_event_position' fields in the common + // header of this event must be set to 0. The 'flags' field in the + // common header must be set to 'artificial'. + // 2. The very first event in the "binlog." (or resolved oldest binlog + // file). It must be the FORMAT_DESCRIPTION event. This event must + // always include checksum (regardless of client or server settings). + // 3. Subsequent events in that "binlog." binlog file. + // 4. The last event in the "binlog." (can be one of the following). + // a. In the most common case it must be a real ROTATE event with the + // 'binlog' field set to "binlog.", a non-zero 'timestamp', + // a non-zero 'next_event_position', and 'flags' field not + // containing the 'artificial' bit. + // b. After MySQL Server shutdown, the last event in a binlog file + // might be a STOP event. + // c. In rare cases, after improper shutdown, the binlog may end simply + // with the last event in a complete transaction (usually XID event). + // 5. A generated artificial ROTATE with 'binlog' field set to + // "binlog." and position set to 4. Whether this event should + // include checksum or not depends on the value of the + // 'checksum_algorithm' field in the last seen FORMAT_DESCRIPTION event, + // the event from (2) in this sequence. + // 6. Similar to (2), but for "binlog.". + // 7. Similar to (3), but for "binlog.". + // 8. Similar to (4), but for "binlog.". + // ... + // N. Repeat steps (5)-(8) for subsequent binlog files. + // If there are no more events in the last binlog file, EOF is returned + // (or, in blocking mode, the caller polls until new events appear). + + // In case when client requests replication from a position that is not + // equal to "magic offset" 4, there are a few changes to the rules + // described above. + // 1. Almost identical to (1), but the 'position' field in the artificial + // ROTATE event must be set to the requested position. Empty binlog name + // is not supported in this case. + // 2. Instead of (2) (a FORMAT_DESCRIPTION event taken from the binlog + // file as is), we must send an artificial FORMAT_DESCRIPTION event. + // It is constructed from the real FORMAT_DESCRIPTION event located at + // the beginning of "binlog." by setting 'next_event_position' field + // in the common header and 'create_timestamp' field in the post header + // to 0 (all the other fields, including 'timestamp' and 'flags' in the + // common header, are kept as is). The checksum is recalculated. + // 3. Events from (3) are sent starting from the requested position. + // These two artificial events, (1) and (2), must be sent even when the + // requested position is equal to the size of "binlog." (meaning that + // there are no more events to send from this file). In this case, if + // "binlog." exists, the sequence continues immediately from step (5) + // of the rules above. Otherwise, EOF is returned. + + // In order to implement this logic, we use the following FSM + // | | + // v | + // (start_from_offset) | + // | | + // v | + // (generate_artificial_fde) | + // | | + // | v + // | (start_from_beginning) <----+ + // | | | + // v v | + // +----> (fetch_event_from_storage) | + // | | | | + // +-------+ +-------------+ + // [ binlog file not changed ] [ binlog file changed ] + // + // 'start_from_offset' and 'generate_artificial_fde' states do not require + // any event block to be fetched from the storage ('start_from_offset' + // reads the FORMAT_DESCRIPTION event from the beginning of "binlog." + // directly). 'start_from_beginning' state, on the other hand, requires + // the first event block to be fetched first, as this is how an empty + // binlog name gets resolved and how switching to the next binlog file + // gets detected. + + // the 'start_from_offset' and 'generate_artificial_fde' states must be + // handled before any event block is fetched: the artificial ROTATE / + // FORMAT_DESCRIPTION pair for the requested binlog file must be sent even + // when the requested position is equal to the size of that file (in which + // case the very first fetch would either return EOF or would switch to the + // next binlog file) + + // a branch with early return for the 'start_from_offset' state + if (fsm_state_ == fsm_state_type::start_from_offset) { + return handle_start_states(event); + } + + // a branch with early return for the 'generate_artificial_fde' state + if (fsm_state_ == fsm_state_type::generate_artificial_fde) { + return handle_generate_artificial_fde_state(event); + } + + const auto populate_result{populate_event_block(event)}; + if (populate_result.has_value()) { + return *populate_result; + } + + // a branch with early return for the 'start_from_beginning' state + if (fsm_state_ == fsm_state_type::start_from_beginning) { + return handle_start_states(event); + } + + // The main branch for the 'fetch_event_from_storage' state + return handle_fetch_event_from_storage_state(event); +} + +[[nodiscard]] std::size_t +sender_context::calculate_fde_size(std::uint32_t encoded_server_version) { + return binsrv::events::default_common_header_length + + binsrv::events::generic_post_header_impl< + binsrv::events::code_type::format_description>:: + get_size_in_bytes(encoded_server_version) + + binsrv::events::generic_body_impl< + binsrv::events::code_type::format_description>::size_in_bytes + + // FORMAT_DESCRIPTION event always has footer + binsrv::events::default_footer_length; +} + +[[nodiscard]] bool sender_context::fetch_fde_from_storage() { + // 'range_' has already been advanced past the fetched event block at this + // point, so the only meaningful invariant here is the FSM state + assert(fsm_state_ == fsm_state_type::start_from_offset); + binsrv::events::composite_binlog_name fde_binlog_name{binlog_name_}; + util::dynamic_byte_buffer fde_buffer{}; + // starting from MySQL Server 8.3 FORMAT_DESCRIPTION event is one byte + // longer because of the new GTID_TAGGED_LOG_EVENT + + // so here we assume that the max buffer size should be large enough + // to hold the entire FORMAT_DESCRIPTION event from the most recent + // known MySQL Server version (>=8.3) + const auto max_fde_size{ + calculate_fde_size(binsrv::events::latest_known_protocol_server_version)}; + util::byte_range fde_range{binsrv::events::magic_binlog_offset, max_fde_size}; + + if (!storage_->fetch_event_block(fde_binlog_name, fde_range, fde_buffer)) { + return false; + } + // the binlog file may be shorter than requested (truncated / corrupt), so + // we need to make sure that at least the common header fits + if (std::size(fde_buffer) < binsrv::events::default_common_header_length) { + return false; + } + const binsrv::events::common_header_view fde_common_header_v{ + util::const_byte_span{fde_buffer}.subspan( + 0, binsrv::events::default_common_header_length)}; + + if (fde_common_header_v.get_type_code() != + binsrv::events::code_type::format_description) { + return false; + } + + const auto min_fde_size{calculate_fde_size( + binsrv::events::earliest_supported_protocol_server_version)}; + const auto real_fde_size{fde_common_header_v.get_event_size_raw()}; + if (real_fde_size < min_fde_size || real_fde_size > max_fde_size) { + return false; + } + // the event size advertised in the common header must not exceed the + // number of bytes actually fetched + if (real_fde_size > std::size(fde_buffer)) { + return false; + } + fde_.assign(std::cbegin(fde_buffer), std::cbegin(fde_buffer) + real_fde_size); + return true; +} + +[[nodiscard]] bool sender_context::extract_fde_from_event_block() { + assert(event_block_); + assert(!event_block_->is_empty()); + const auto fde_span{event_block_->get_event(0U)}; + + const binsrv::events::common_header_view fde_common_header_v{ + util::const_byte_span{fde_span}.subspan( + 0, binsrv::events::default_common_header_length)}; + if (fde_common_header_v.get_type_code() != + binsrv::events::code_type::format_description) { + return false; + } + fde_.assign(std::cbegin(fde_span), std::cend(fde_span)); + return true; +} + +// NOLINTNEXTLINE(bugprone-easily-swappable-parameters) +void sender_context::extract_fields_from_fde(std::uint32_t &server_id, + bool &checksum_enabled) const { + // creating a minimally-initialized reader context with only one + // meaningful field 'checksum_verification_enabled' set to true - + // creating a view on an FDE is a special case and will not read + // any other field from the context + const binsrv::events::reader_context fake_ctx{ + 0U, /* connection_encoded_server_version */ + true, /* checksum_verification_enabled */ + binsrv::replication_mode_type::position, /* replication_mode */ + "", /* binlog_name */ + 0U, /* position */ + }; + const util::const_byte_span fde_span{fde_}; + const binsrv::events::event_view fde_v{fake_ctx, fde_span}; + + const auto fde_common_header_v{fde_v.get_common_header_view()}; + + assert(fde_common_header_v.get_type_code() == + binsrv::events::code_type::format_description); + + const binsrv::events::generic_body_impl< + binsrv::events::code_type::format_description> + fde_body{fde_v.get_body_raw()}; + + server_id = fde_common_header_v.get_server_id_raw(); + checksum_enabled = (fde_body.get_checksum_algorithm() == + binsrv::events::checksum_algorithm_type::crc32); +} + +void sender_context::transform_fde_to_artificial() { + // artificial FDE must have 'next_event_pos' field in the common header + // set to 0U and 'create_timestamp' in the post header set to 0U + const binsrv::events::reader_context fake_ctx{ + 0U, /* connection_encoded_server_version */ + true, /* checksum_verification_enabled */ + binsrv::replication_mode_type::position, /* replication_mode */ + "", /* binlog_name */ + 0U, /* position */ + }; + const util::byte_span fde_span{fde_}; + const binsrv::events::event_updatable_view fde_uv{fake_ctx, fde_span}; + { + const auto write_proxy{fde_uv.get_write_proxy()}; + const auto fde_common_header_uv{ + write_proxy.get_common_header_updatable_view()}; + fde_common_header_uv.set_next_event_position_raw(0U); + auto fde_post_header_span{write_proxy.get_post_header_updatable_raw()}; + binsrv::events::generic_post_header_impl< + binsrv::events::code_type::format_description> + fde_post_header{fde_post_header_span}; + fde_post_header.set_create_timestamp_raw(0U); + fde_post_header.encode_to(fde_post_header_span); + } +} + +[[nodiscard]] util::optional_bool +sender_context::populate_event_block(util::const_byte_span &event) { + if (event_block_ && event_index_ != event_block_->get_number_of_events()) { + return {}; + } + // if this is the very first call when 'event_block_' is not yet set or we // have consumed all events in the current block - if (!event_block_ || event_index_ == event_block_->get_number_of_events()) { - // early reset to free memory from the previous event block + + // early reset to free memory from the previous event block + event_block_.reset(); + + util::dynamic_byte_buffer buffer{}; + const binsrv::events::composite_binlog_name saved_binlog_name{binlog_name_}; + + // on success both 'binlog_name_' and 'range_' will be updated + if (!storage_->fetch_event_block(binlog_name_, range_, buffer)) { + return false; + } + + if (saved_binlog_name.is_empty()) { + logger_->log_format(binsrv::log_severity::info, + "sender : empty binlog name resolved to {}", + binlog_name_.str()); + } else { + if (binlog_name_ != saved_binlog_name) { + logger_->log_format(binsrv::log_severity::info, + "sender : switched to a new binlog file {} -> {}", + saved_binlog_name.str(), binlog_name_.str()); + fsm_state_ = fsm_state_type::start_from_beginning; + } + } + if (buffer.empty()) { + logger_->log(binsrv::log_severity::info, "sender : fetched EOF"); + // leaving 'binlog_name_' as is so that on next fetch after this + // EOF we could make another attempt to check if new events were added + + // 'range_', on the other hand, was set to empty inside + // the 'fetch_event_block()' - here we restore its length for the + // next fetch attempt + range_ = util::byte_range{range_.get_offset(), block_size_}; event_block_.reset(); + event_index_ = 0UZ; - util::dynamic_byte_buffer buffer{}; - // on success both 'binlog_name_' and 'range_' will be updated + // setting the event span to an empty object to indicate EOF + event = util::const_byte_span{}; + return true; // EOF + } + logger_->log_format(binsrv::log_severity::info, + "sender : fetched event block of size {}, {}:{}", + std::size(buffer), binlog_name_.str(), + range_.to_string()); + + // Check that the buffer contains at least one complete event before + // parsing. If the first event's common header does not even fit, the + // binlog is corrupt or truncated. If the header fits but advertises an + // event larger than what we fetched, re-issue the fetch at the same + // offset with exactly 'event_size' bytes. + if (std::size(buffer) < + binsrv::events::common_header_view_base::size_in_bytes) { + return false; + } + const binsrv::events::common_header_view header{ + util::const_byte_span{buffer}.subspan( + 0UZ, binsrv::events::common_header_view_base::size_in_bytes)}; + const auto first_event_size{ + static_cast(header.get_event_size_raw())}; + if (first_event_size > binsrv::events::max_event_size_bytes) { + // Nonsensically large event size: corrupt or malicious header. + return false; + } + if (first_event_size > std::size(buffer)) { + logger_->log_format( + binsrv::log_severity::info, + "sender : block too small for first event (needs {} bytes), " + "retrying with exact size at {}:{}", + first_event_size, binlog_name_.str(), range_.to_string()); + + buffer.clear(); + range_ = util::byte_range{range_.get_offset(), first_event_size}; if (!storage_->fetch_event_block(binlog_name_, range_, buffer)) { return false; } - if (buffer.empty()) { - logger_->log(binsrv::log_severity::info, "sender : fetched EOF"); - // On EOF, 'storage::fetch_event_block()' leaves 'range_' with length 0, - // and unchanged offset. Restore the length while - // keeping 'binlog_name_' and the current offset so that a subsequent - // call resumes polling at the same position. - range_ = util::byte_range{range_.get_offset(), block_size_}; - - // setting the event span to an empty object to indicate EOF - event = util::const_byte_span{}; - return true; // EOF - } - logger_->log_format(binsrv::log_severity::info, - "sender : fetched event block of size {}, {}:{}", - std::size(buffer), binlog_name_.str(), - range_.to_string()); - - // Check that the buffer contains at least one complete event before - // parsing. If the first event's common header does not even fit, the - // binlog is corrupt or truncated. If the header fits but advertises an - // event larger than what we fetched, re-issue the fetch at the same - // offset with exactly 'event_size' bytes. - if (std::size(buffer) < - binsrv::events::common_header_view_base::size_in_bytes) { + } + + event_block_ = + std::make_unique(std::move(buffer)); + if (event_block_->is_empty()) { + // Exact-size request still yielded no complete event: corrupt binlog. + return false; + } + event_index_ = 0UZ; + logger_->log_format( + binsrv::log_severity::info, + "sender : parsed event block with {} events, actual size {} byte(s)", + event_block_->get_number_of_events(), event_block_->get_actual_size()); + // preparing 'range_' for the next fetch + range_ = util::byte_range{ + range_.get_offset() + event_block_->get_actual_size(), block_size_}; + return {}; +} + +[[nodiscard]] bool +sender_context::handle_start_states(util::const_byte_span &event) { + assert(fsm_state_ == fsm_state_type::start_from_beginning || + fsm_state_ == fsm_state_type::start_from_offset); + // in case when we start from the 'start_from_offset' state, in addition to + // fetching real events requested in the range, we also need to receive + // FORMAT_DESCRIPTION event located at the beginning of this binlog file + if (fsm_state_ == fsm_state_type::start_from_offset) { + // as the 'start_from_offset' state is handled before any event block is + // fetched, an empty binlog name has not been resolved yet - starting from + // a non-magic offset of an unspecified binlog file is not supported + if (binlog_name_.is_empty()) { + logger_->log(binsrv::log_severity::error, + "sender : cannot start from a position other than 4 without " + "specifying binlog name"); return false; } - const binsrv::events::common_header_view header{ - util::const_byte_span{buffer}.subspan( - 0UZ, binsrv::events::common_header_view_base::size_in_bytes)}; - const auto first_event_size{ - static_cast(header.get_event_size_raw())}; - if (first_event_size > binsrv::events::max_event_size_bytes) { - // Nonsensically large event size: corrupt or malicious header. + if (!fetch_fde_from_storage()) { return false; } - if (first_event_size > std::size(buffer)) { - logger_->log_format( - binsrv::log_severity::info, - "sender : block too small for first event (needs {} bytes), " - "retrying with exact size at {}:{}", - first_event_size, binlog_name_.str(), range_.to_string()); - - buffer.clear(); - range_ = util::byte_range{range_.get_offset(), first_event_size}; - if (!storage_->fetch_event_block(binlog_name_, range_, buffer)) { - return false; - } - } - - event_block_ = - std::make_unique(std::move(buffer)); - if (event_block_->is_empty()) { - // Exact-size request still yielded no complete event: corrupt binlog. + } else { + if (!extract_fde_from_event_block()) { return false; } - event_index_ = 0UZ; - logger_->log_format( - binsrv::log_severity::info, - "sender : parsed event block with {} events, actual size {} byte(s)", - event_block_->get_number_of_events(), event_block_->get_actual_size()); - // preparing 'range_' for the next fetch - range_ = util::byte_range{ - range_.get_offset() + event_block_->get_actual_size(), block_size_}; } + + std::uint32_t fde_server_id{}; + bool fde_checksum_enabled{}; + extract_fields_from_fde(fde_server_id, fde_checksum_enabled); + + // for artificial events it is OK to pass magic_binlog_offset as the + // offset as it will be ignored anyway + generate_rotate_event_ex(artificial_rotate_, current_binlog_checksum_, + binsrv::events::magic_binlog_offset, + false /* zero timestamp */, fde_server_id, + true /* artificial */, binlog_name_, + position_for_artificial_rotate_); + position_for_artificial_rotate_ = binsrv::events::magic_binlog_offset; + + // for the artificial ROTATE events generated in future the decision + // whether to include footer with a checksum will be based on the + // 'checksum_algorithm' field from the most recent FDE + current_binlog_checksum_ = fde_checksum_enabled; + + fsm_state_ = fsm_state_ == fsm_state_type::start_from_beginning + ? fsm_state_type::fetch_event_from_storage + : fsm_state_type::generate_artificial_fde; + + event = artificial_rotate_; + return true; +} + +[[nodiscard]] bool sender_context::handle_generate_artificial_fde_state( + util::const_byte_span &event) { + assert(fsm_state_ == fsm_state_type::generate_artificial_fde); + + // at this point 'fde_' must have already been filled with the first + // FDE in the current binlog file + transform_fde_to_artificial(); + fsm_state_ = fsm_state_type::fetch_event_from_storage; + + event = fde_; + return true; +} + +[[nodiscard]] bool sender_context::handle_fetch_event_from_storage_state( + util::const_byte_span &event) { + assert(fsm_state_ == fsm_state_type::fetch_event_from_storage); + event = event_block_->get_event(event_index_); ++event_index_; return true; diff --git a/src/operations/sender_context.hpp b/src/operations/sender_context.hpp index 8a105b9e..912597c5 100644 --- a/src/operations/sender_context.hpp +++ b/src/operations/sender_context.hpp @@ -19,15 +19,21 @@ #include "operations/sender_context_fwd.hpp" // IWYU pragma: export #include +#include + +// needed for 'event_storage' +#include // IWYU pragma: keep #include "binsrv/basic_logger_fwd.hpp" #include "binsrv/indexed_event_block_fwd.hpp" #include "binsrv/storage_fwd.hpp" #include "binsrv/events/composite_binlog_name.hpp" +#include "binsrv/events/event_fwd.hpp" #include "util/byte_range.hpp" #include "util/byte_span_fwd.hpp" +#include "util/common_optional_types.hpp" namespace operations { @@ -35,7 +41,8 @@ class sender_context { public: // deliberately passing by value as we will be moving from these objects sender_context(binsrv::basic_logger_ptr logger, binsrv::storage_ptr storage, - std::size_t block_size); + std::size_t block_size, std::string_view binlog_name, + std::uint64_t position, bool session_source_binlog_checksum); sender_context(const sender_context &) = delete; sender_context &operator=(const sender_context &) = delete; @@ -53,10 +60,39 @@ class sender_context { binsrv::storage_ptr storage_{}; std::size_t block_size_{}; + bool current_binlog_checksum_{}; + + enum class fsm_state_type : std::uint8_t { + start_from_beginning, + start_from_offset, + generate_artificial_fde, + fetch_event_from_storage + }; + fsm_state_type fsm_state_{}; + + std::uint64_t position_for_artificial_rotate_{}; binsrv::events::composite_binlog_name binlog_name_{}; util::byte_range range_{}; + binsrv::events::event_storage artificial_rotate_{}; + binsrv::events::event_storage fde_{}; binsrv::indexed_event_block_ptr event_block_{}; std::size_t event_index_{}; + + [[nodiscard]] static std::size_t + calculate_fde_size(std::uint32_t encoded_server_version); + [[nodiscard]] bool fetch_fde_from_storage(); + [[nodiscard]] bool extract_fde_from_event_block(); + void extract_fields_from_fde(std::uint32_t &server_id, + bool &checksum_enabled) const; + void transform_fde_to_artificial(); + + [[nodiscard]] util::optional_bool + populate_event_block(util::const_byte_span &event); + [[nodiscard]] bool handle_start_states(util::const_byte_span &event); + [[nodiscard]] bool + handle_generate_artificial_fde_state(util::const_byte_span &event); + [[nodiscard]] bool + handle_fetch_event_from_storage_state(util::const_byte_span &event); }; } // namespace operations diff --git a/src/util/common_optional_types.hpp b/src/util/common_optional_types.hpp index dacad630..4c36136b 100644 --- a/src/util/common_optional_types.hpp +++ b/src/util/common_optional_types.hpp @@ -24,6 +24,8 @@ namespace util { using optional_string = std::optional; +using optional_bool = std::optional; + using optional_uint8_t = std::optional; using optional_uint16_t = std::optional; using optional_uint32_t = std::optional; diff --git a/src/util/flag_set.hpp b/src/util/flag_set.hpp index 7dfba7c5..9c27a907 100644 --- a/src/util/flag_set.hpp +++ b/src/util/flag_set.hpp @@ -53,6 +53,11 @@ class [[nodiscard]] flag_set { assert(std::has_single_bit(underlying)); bits_ |= underlying; } + void clear_element(element_type element) noexcept { + auto underlying{static_cast(element)}; + assert(std::has_single_bit(underlying)); + bits_ &= static_cast(~underlying); + } void flip_element(element_type element) noexcept { auto underlying{static_cast(element)}; assert(std::has_single_bit(underlying)); From fad1d58b6473ff41416e577a9542bfbe7170e3f5 Mon Sep 17 00:00:00 2001 From: Yura Sorokin Date: Wed, 7 Oct 2026 23:33:58 +0200 Subject: [PATCH 2/2] PBS-42 feature: Implement basic support for COM_BINLOG_DUMP packet handling (part 6) https://perconadev.atlassian.net/browse/PBS-42 Added new 'binlog_streaming.rs_position_based_consistency' MTR test case. It generates several binlog files (including an empty one), records (binlog name, position) checkpoints in a JSON session variable after each step, and also adds ["":4] and the end-of-file position of every binlog file. For each checkpoint 'mysqlbinlog --to-last-log' is run both against the PBS and directly against the MySQL Server and the generated event sequences are compared. The test case is run with the same 4 combinations as 'binlog_streaming.rs_roundtrip'. Extracted dump-and-compare logic into the new 'rs_compare_dumps.inc' include file and reused it in 'rs_roundtrip_body.inc' (shared by 'binlog_streaming.rs_roundtrip' and 'binlog_streaming.rs_roundtrip_encryption'). Co-Authored-By: Claude Opus 5.5 (1M context) --- .../include/rs_compare_dumps.inc | 61 +++++++ .../rs_position_based_consistency_save.inc | 40 +++++ .../include/rs_roundtrip_body.inc | 69 ++------ .../r/rs_position_based_consistency.result | 106 ++++++++++++ mtr/binlog_streaming/r/rs_roundtrip.result | 20 +-- .../r/rs_roundtrip_encryption.result | 20 +-- ...rs_position_based_consistency.combinations | 13 ++ .../t/rs_position_based_consistency.test | 156 ++++++++++++++++++ 8 files changed, 402 insertions(+), 83 deletions(-) create mode 100644 mtr/binlog_streaming/include/rs_compare_dumps.inc create mode 100644 mtr/binlog_streaming/include/rs_position_based_consistency_save.inc create mode 100644 mtr/binlog_streaming/r/rs_position_based_consistency.result create mode 100644 mtr/binlog_streaming/t/rs_position_based_consistency.combinations create mode 100644 mtr/binlog_streaming/t/rs_position_based_consistency.test diff --git a/mtr/binlog_streaming/include/rs_compare_dumps.inc b/mtr/binlog_streaming/include/rs_compare_dumps.inc new file mode 100644 index 00000000..556bbd60 --- /dev/null +++ b/mtr/binlog_streaming/include/rs_compare_dumps.inc @@ -0,0 +1,61 @@ +# ==== Purpose ==== +# +# Dumps binary logs starting from the specified (binlog name, position) +# checkpoint up to the last binlog file both from the Binlog Server +# replication source and directly from the MySQL Server and compares +# generated event sequences. +# +# ==== Usage ==== +# +# --let $checkpoint_label = after inserting the first batch of rows +# --let $checkpoint_binlog_name = binlog.000001 +# --let $checkpoint_binlog_position = 4 +# [--let $saved_roundtrip_dump = $MYSQL_TMP_DIR/roundtrip.sql] +# --source ../include/rs_compare_dumps.inc +# +# An empty $checkpoint_binlog_name means "start from the oldest binlog file +# available" (in this case $checkpoint_binlog_position must be 4). +# +# If $saved_roundtrip_dump is specified, the unfiltered dump received from +# the Binlog Server replication source is copied to this location before +# the cleanup (e.g. to be restored later). It is the responsibility of the +# caller to remove this file. $saved_roundtrip_dump is reset at the end. +# +# Expected inputs set by set_up_binsrv_environment.inc: +# $binsrv_connection_host +# $binsrv_connection_user +# $binsrv_replication_source_port +# $binsrv_auth_user +# $binsrv_auth_password + +--echo +--echo *** Comparing event sequences from the PBS and MySQL Server starting +--echo *** from checkpoint '$checkpoint_label'. +--let $checkpoint_roundtrip_dump = $MYSQL_TMP_DIR/rs_compare_dumps_roundtrip.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port + --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password + --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $checkpoint_roundtrip_dump 2>/dev/null; +--let $checkpoint_roundtrip_dump_filtered = $checkpoint_roundtrip_dump.filtered +--exec cat $checkpoint_roundtrip_dump | grep -E "^#[[:digit:]]+" > $checkpoint_roundtrip_dump_filtered + +--let $checkpoint_direct_dump = $MYSQL_TMP_DIR/rs_compare_dumps_direct.sql +exec $MYSQL_BINLOG + --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user + --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $checkpoint_direct_dump 2>/dev/null; +--let $checkpoint_direct_dump_filtered = $checkpoint_direct_dump.filtered +--exec cat $checkpoint_direct_dump | grep -E "^#[[:digit:]]+" > $checkpoint_direct_dump_filtered + +--diff_files $checkpoint_roundtrip_dump_filtered $checkpoint_direct_dump_filtered + +if ($saved_roundtrip_dump) +{ + --copy_file $checkpoint_roundtrip_dump $saved_roundtrip_dump +} + +--remove_file $checkpoint_roundtrip_dump +--remove_file $checkpoint_roundtrip_dump_filtered +--remove_file $checkpoint_direct_dump +--remove_file $checkpoint_direct_dump_filtered + +--let $saved_roundtrip_dump = diff --git a/mtr/binlog_streaming/include/rs_position_based_consistency_save.inc b/mtr/binlog_streaming/include/rs_position_based_consistency_save.inc new file mode 100644 index 00000000..8ca14fb6 --- /dev/null +++ b/mtr/binlog_streaming/include/rs_position_based_consistency_save.inc @@ -0,0 +1,40 @@ +# ==== Purpose ==== +# +# Appends a (binlog name, position) checkpoint to the '@checkpoints' +# JSON array session variable initialized by +# 'rs_position_based_consistency.test'. Setting a user variable does not +# write anything to the binary log, so saving a checkpoint does not affect +# the content of binary log files. +# +# ==== Usage ==== +# +# --let $checkpoint_label = after inserting the first batch of rows +# [--let $checkpoint_binlog_name = binlog.000001] +# [--let $checkpoint_binlog_position = 4] +# --source ../include/rs_position_based_consistency_save.inc +# +# If $checkpoint_binlog_position is not specified, the current binary log +# name and position are taken from $stmt_show_binary_log_status (set by +# v80_v84_compatibility_defines.inc). If $checkpoint_binlog_position is +# specified, $checkpoint_binlog_name is used as is (an empty +# $checkpoint_binlog_name with position 4 means "from the beginning of the +# oldest binlog file"). +# +# The $checkpoint_label must not contain single quotes. + +if ($checkpoint_binlog_position == "") +{ + --let $checkpoint_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1) + --let $checkpoint_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1) +} + +--disable_query_log +eval SET @checkpoints = JSON_ARRAY_APPEND(@checkpoints, '\$', + JSON_OBJECT('binlog_file', '$checkpoint_binlog_name', + 'position', $checkpoint_binlog_position, + 'test_label', '$checkpoint_label')); +--enable_query_log + +--let $checkpoint_binlog_name = +--let $checkpoint_binlog_position = +--let $checkpoint_label = diff --git a/mtr/binlog_streaming/include/rs_roundtrip_body.inc b/mtr/binlog_streaming/include/rs_roundtrip_body.inc index de1d80d6..ad927ada 100644 --- a/mtr/binlog_streaming/include/rs_roundtrip_body.inc +++ b/mtr/binlog_streaming/include/rs_roundtrip_body.inc @@ -33,9 +33,9 @@ FLUSH BINARY LOGS; INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); --echo ---echo *** Determining checkpoint binary log name and position. ---let $checkpoint_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1) ---let $checkpoint_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1) +--echo *** Determining binary log name and position for the second part. +--let $second_part_binlog_name = query_get_value($stmt_show_binary_log_status, File, 1) +--let $second_part_binlog_position = query_get_value($stmt_show_binary_log_status, Position, 1) --echo --echo *** Fetching the source binary logs in position mode for the first time. @@ -52,29 +52,12 @@ INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); --let $listening_port = $binsrv_replication_source_port --source ../include/wait_for_listening_port.inc ---echo ---echo *** Dumping the first part of binary logs from the Binlog Server ---echo *** replication source. --let $first_roundtrip_dump = $MYSQL_TMP_DIR/first_rs_roundtrip.sql -exec $MYSQL_BINLOG - --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port - --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password - --to-last-log "" > $first_roundtrip_dump 2>/dev/null; ---let $first_roundtrip_dump_filtered = $first_roundtrip_dump.filtered ---exec cat $first_roundtrip_dump | grep -E "^#[[:digit:]]+" > $first_roundtrip_dump_filtered - ---echo ---echo *** Dumping the first part of binary logs from the MySQL Server ---let $first_direct_dump = $MYSQL_TMP_DIR/first_rs_direct.sql -exec $MYSQL_BINLOG - --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user - --to-last-log "" > $first_direct_dump 2>/dev/null; ---let $first_direct_dump_filtered = $first_direct_dump.filtered ---exec cat $first_direct_dump | grep -E "^#[[:digit:]]+" > $first_direct_dump_filtered - ---echo ---echo *** Comparing the first parts of binary logs from the PBS and MySQL Server ---diff_files $first_roundtrip_dump_filtered $first_direct_dump_filtered +--let $checkpoint_label = the beginning of the oldest binlog file +--let $checkpoint_binlog_name = +--let $checkpoint_binlog_position = 4 +--let $saved_roundtrip_dump = $first_roundtrip_dump +--source ../include/rs_compare_dumps.inc --echo --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for @@ -83,7 +66,7 @@ exec $MYSQL_BINLOG --echo --echo *** Inserting the third batch of rows. -INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine'); --echo --echo *** Flushing the source binary log for the second time. @@ -107,29 +90,12 @@ INSERT INTO source_db.tbl VALUES (10, 'ten'), (11, 'eleven'), (12, 'twelve'); --let $listening_port = $binsrv_replication_source_port --source ../include/wait_for_listening_port.inc ---echo ---echo *** Dumping the second part of binary logs from the Binlog Server ---echo *** replication source. --let $second_roundtrip_dump = $MYSQL_TMP_DIR/second_rs_roundtrip.sql -exec $MYSQL_BINLOG - --read-from-remote-server --host=$binsrv_connection_host --port=$binsrv_replication_source_port - --user=$binsrv_auth_user --password=$binsrv_auth_password --default-auth=caching_sha2_password - --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $second_roundtrip_dump 2>/dev/null; ---let $second_roundtrip_dump_filtered = $second_roundtrip_dump.filtered ---exec cat $second_roundtrip_dump | grep -E "^#[[:digit:]]+" > $second_roundtrip_dump_filtered - ---echo ---echo *** Dumping the second part of binary logs from the MySQL Server ---let $second_direct_dump = $MYSQL_TMP_DIR/second_rs_direct.sql -exec $MYSQL_BINLOG - --read-from-remote-server --host=$binsrv_connection_host --user=$binsrv_connection_user - --start-position=$checkpoint_binlog_position --to-last-log "$checkpoint_binlog_name" > $second_direct_dump 2>/dev/null; ---let $second_direct_dump_filtered = $second_direct_dump.filtered ---exec cat $second_direct_dump | grep -E "^#[[:digit:]]+" > $second_direct_dump_filtered - ---echo ---echo *** Comparing the second parts of binary logs from the PBS and MySQL Server ---diff_files $second_roundtrip_dump_filtered $second_direct_dump_filtered +--let $checkpoint_label = after inserting the second batch of rows +--let $checkpoint_binlog_name = $second_part_binlog_name +--let $checkpoint_binlog_position = $second_part_binlog_position +--let $saved_roundtrip_dump = $second_roundtrip_dump +--source ../include/rs_compare_dumps.inc --echo --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for @@ -160,11 +126,4 @@ DROP DATABASE backup_db; --echo --echo *** Cleaning up. --remove_file $first_roundtrip_dump ---remove_file $first_roundtrip_dump_filtered ---remove_file $first_direct_dump ---remove_file $first_direct_dump_filtered - --remove_file $second_roundtrip_dump ---remove_file $second_roundtrip_dump_filtered ---remove_file $second_direct_dump ---remove_file $second_direct_dump_filtered diff --git a/mtr/binlog_streaming/r/rs_position_based_consistency.result b/mtr/binlog_streaming/r/rs_position_based_consistency.result new file mode 100644 index 00000000..4f4d012a --- /dev/null +++ b/mtr/binlog_streaming/r/rs_position_based_consistency.result @@ -0,0 +1,106 @@ +*** Resetting replication at the very beginning of the test. + +*** Generating a configuration file in JSON format for the Binlog +*** Server utility. + +*** Determining binlog file directory from the server. + +*** Creating a temporary directory for storing +*** binlog files downloaded via the Binlog Server utility. + +*** Initializing the @checkpoints session variable (a JSON array) for +*** storing (binlog name, position) checkpoints. + +*** Saving the ["":4] checkpoint (everything from the beginning of the +*** oldest binary log file). + +*** Step 0: Saving the initial binary log position. + +*** Step 1: Creating tbl. +CREATE TABLE tbl( +id INT UNSIGNED NOT NULL, +value VARCHAR(32) NOT NULL, +PRIMARY KEY(id) +) ENGINE=InnoDB; + +*** Step 2: Inserting the first batch of rows. +INSERT INTO tbl VALUES (1, 'one'), (2, 'two'), (3, 'three'); + +*** Step 3: Flushing the source binary log for the first time. +FLUSH BINARY LOGS; + +*** Step 4: Inserting the second batch of rows. +INSERT INTO tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); + +*** Step 5: Flushing the source binary log for the second time. +FLUSH BINARY LOGS; + +*** Step 6: Flushing the source binary log for the third time (producing an +*** empty binary log file). +FLUSH BINARY LOGS; + +*** Step 7: Inserting the third batch of rows. +INSERT INTO tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine'); + +*** Saving end-of-file positions (sizes) of all binary log files as +*** checkpoints. +include/assert.inc [There must be exactly 4 binary log files] + +*** Fetching the source binary logs in position mode. + +*** Starting Binlog Server Utility in background in pull mode. +include/read_file_to_var.inc + +*** Waiting for the replication source listener port ready to accept +*** connections. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'from the beginning of the oldest binary log file'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'at the very beginning'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after creating tbl'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after inserting the first batch of rows'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after the first FLUSH BINARY LOGS'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after inserting the second batch of rows'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after the second FLUSH BINARY LOGS'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after the third FLUSH BINARY LOGS'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after inserting the third batch of rows'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'end of binary log file 1'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'end of binary log file 2'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'end of binary log file 3'. + +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'end of binary log file 4'. + +*** Sending SIGTERM signal to the Binlog Server Utility and waiting for +*** the process to terminate. + +*** Dropping the table. +DROP TABLE tbl; + +*** Removing the Binlog Server utility storage directory. + +*** Removing the Binlog Server utility log file. + +*** Removing the Binlog Server utility configuration file. diff --git a/mtr/binlog_streaming/r/rs_roundtrip.result b/mtr/binlog_streaming/r/rs_roundtrip.result index 574fa97b..e4c86803 100644 --- a/mtr/binlog_streaming/r/rs_roundtrip.result +++ b/mtr/binlog_streaming/r/rs_roundtrip.result @@ -23,7 +23,7 @@ FLUSH BINARY LOGS; *** Inserting the second batch of rows. INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); -*** Determining checkpoint binary log name and position. +*** Determining binary log name and position for the second part. *** Fetching the source binary logs in position mode for the first time. @@ -33,18 +33,14 @@ include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept *** connections for the first time. -*** Dumping the first part of binary logs from the Binlog Server -*** replication source. - -*** Dumping the first part of binary logs from the MySQL Server - -*** Comparing the first parts of binary logs from the PBS and MySQL Server +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'the beginning of the oldest binlog file'. *** Sending SIGTERM signal to the Binlog Server Utility and waiting for *** the process to terminate for the first time. *** Inserting the third batch of rows. -INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine'); *** Flushing the source binary log for the second time. FLUSH BINARY LOGS; @@ -59,12 +55,8 @@ include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept *** connections for the second time. -*** Dumping the second part of binary logs from the Binlog Server -*** replication source. - -*** Dumping the second part of binary logs from the MySQL Server - -*** Comparing the second parts of binary logs from the PBS and MySQL Server +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after inserting the second batch of rows'. *** Sending SIGTERM signal to the Binlog Server Utility and waiting for *** the process to terminate for the second time. diff --git a/mtr/binlog_streaming/r/rs_roundtrip_encryption.result b/mtr/binlog_streaming/r/rs_roundtrip_encryption.result index 74b7fe2c..6973d6a5 100644 --- a/mtr/binlog_streaming/r/rs_roundtrip_encryption.result +++ b/mtr/binlog_streaming/r/rs_roundtrip_encryption.result @@ -25,7 +25,7 @@ FLUSH BINARY LOGS; *** Inserting the second batch of rows. INSERT INTO source_db.tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); -*** Determining checkpoint binary log name and position. +*** Determining binary log name and position for the second part. *** Fetching the source binary logs in position mode for the first time. @@ -35,18 +35,14 @@ include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept *** connections for the first time. -*** Dumping the first part of binary logs from the Binlog Server -*** replication source. - -*** Dumping the first part of binary logs from the MySQL Server - -*** Comparing the first parts of binary logs from the PBS and MySQL Server +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'the beginning of the oldest binlog file'. *** Sending SIGTERM signal to the Binlog Server Utility and waiting for *** the process to terminate for the first time. *** Inserting the third batch of rows. -INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'none'); +INSERT INTO source_db.tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine'); *** Flushing the source binary log for the second time. FLUSH BINARY LOGS; @@ -61,12 +57,8 @@ include/read_file_to_var.inc *** Waiting for the replication source listener port ready to accept *** connections for the second time. -*** Dumping the second part of binary logs from the Binlog Server -*** replication source. - -*** Dumping the second part of binary logs from the MySQL Server - -*** Comparing the second parts of binary logs from the PBS and MySQL Server +*** Comparing event sequences from the PBS and MySQL Server starting +*** from checkpoint 'after inserting the second batch of rows'. *** Sending SIGTERM signal to the Binlog Server Utility and waiting for *** the process to terminate for the second time. diff --git a/mtr/binlog_streaming/t/rs_position_based_consistency.combinations b/mtr/binlog_streaming/t/rs_position_based_consistency.combinations new file mode 100644 index 00000000..6d42b428 --- /dev/null +++ b/mtr/binlog_streaming/t/rs_position_based_consistency.combinations @@ -0,0 +1,13 @@ +[srv_nocrc_default_id] +binlog-checksum = NONE + +[srv_nocrc_custom_id] +binlog-checksum = NONE +server-id = 67 + +[srv_crc_default_id] +binlog-checksum = CRC32 + +[srv_crc_custom_id] +binlog-checksum = CRC32 +server-id = 67 diff --git a/mtr/binlog_streaming/t/rs_position_based_consistency.test b/mtr/binlog_streaming/t/rs_position_based_consistency.test new file mode 100644 index 00000000..c279fdd7 --- /dev/null +++ b/mtr/binlog_streaming/t/rs_position_based_consistency.test @@ -0,0 +1,156 @@ +--source ../include/have_binsrv.inc + +--source ../include/v80_v84_compatibility_defines.inc + +--source include/count_sessions.inc + +--echo *** Resetting replication at the very beginning of the test. +--disable_query_log +eval $stmt_reset_binary_logs_and_gtids; +--enable_query_log + +--source ../include/identify_storage_backend.inc + +--let $binsrv_connect_timeout = 3 +--let $binsrv_read_timeout = 3 +--let $binsrv_idle_time = 1 +--let $binsrv_verify_checksum = TRUE +--let $binsrv_replication_mode = position +--let $binsrv_checkpoint_size = 1 +--source ../include/set_up_binsrv_environment.inc + +--echo +--echo *** Initializing the @checkpoints session variable (a JSON array) for +--echo *** storing (binlog name, position) checkpoints. +--disable_query_log +SET @checkpoints = JSON_ARRAY(); +--enable_query_log + +--echo +--echo *** Saving the ["":4] checkpoint (everything from the beginning of the +--echo *** oldest binary log file). +--let $checkpoint_label = from the beginning of the oldest binary log file +--let $checkpoint_binlog_name = +--let $checkpoint_binlog_position = 4 +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 0: Saving the initial binary log position. +--let $checkpoint_label = at the very beginning +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 1: Creating tbl. +CREATE TABLE tbl( + id INT UNSIGNED NOT NULL, + value VARCHAR(32) NOT NULL, + PRIMARY KEY(id) +) ENGINE=InnoDB; +--let $checkpoint_label = after creating tbl +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 2: Inserting the first batch of rows. +INSERT INTO tbl VALUES (1, 'one'), (2, 'two'), (3, 'three'); +--let $checkpoint_label = after inserting the first batch of rows +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 3: Flushing the source binary log for the first time. +FLUSH BINARY LOGS; +--let $checkpoint_label = after the first FLUSH BINARY LOGS +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 4: Inserting the second batch of rows. +INSERT INTO tbl VALUES (4, 'four'), (5, 'five'), (6, 'six'); +--let $checkpoint_label = after inserting the second batch of rows +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 5: Flushing the source binary log for the second time. +FLUSH BINARY LOGS; +--let $checkpoint_label = after the second FLUSH BINARY LOGS +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 6: Flushing the source binary log for the third time (producing an +--echo *** empty binary log file). +FLUSH BINARY LOGS; +--let $checkpoint_label = after the third FLUSH BINARY LOGS +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Step 7: Inserting the third batch of rows. +INSERT INTO tbl VALUES (7, 'seven'), (8, 'eight'), (9, 'nine'); +--let $checkpoint_label = after inserting the third batch of rows +--source ../include/rs_position_based_consistency_save.inc + +--echo +--echo *** Saving end-of-file positions (sizes) of all binary log files as +--echo *** checkpoints. +--let $binlog_file_index = 1 +--let $binlog_file_name = query_get_value(SHOW BINARY LOGS, Log_name, $binlog_file_index) +--let $binlog_file_exists = `SELECT '$binlog_file_name' <> 'No such row'` +while ($binlog_file_exists) +{ + --let $checkpoint_binlog_position = query_get_value(SHOW BINARY LOGS, File_size, $binlog_file_index) + --let $checkpoint_binlog_name = $binlog_file_name + --let $checkpoint_label = end of binary log file $binlog_file_index + --source ../include/rs_position_based_consistency_save.inc + --inc $binlog_file_index + --let $binlog_file_name = query_get_value(SHOW BINARY LOGS, Log_name, $binlog_file_index) + --let $binlog_file_exists = `SELECT '$binlog_file_name' <> 'No such row'` +} +--let $assert_text = There must be exactly 4 binary log files +--let $assert_cond = $binlog_file_index = 5 +--source include/assert.inc + +--echo +--echo *** Fetching the source binary logs in position mode. +--exec $BINSRV fetch $binsrv_config_file_path > /dev/null + +--echo +--echo *** Starting Binlog Server Utility in background in pull mode. +--let $proc_command_line = $BINSRV pull $binsrv_config_file_path > /dev/null +--source ../include/start_proc_in_background.inc + +--echo +--echo *** Waiting for the replication source listener port ready to accept +--echo *** connections. +--let $listening_port = $binsrv_replication_source_port +--source ../include/wait_for_listening_port.inc + +--let $number_of_checkpoints = `SELECT JSON_LENGTH(@checkpoints)` +--let $checkpoint_json_index = 0 +while ($checkpoint_json_index < $number_of_checkpoints) +{ + --let $checkpoint_binlog_name = `SELECT JSON_UNQUOTE(JSON_EXTRACT(@checkpoints, '\$[$checkpoint_json_index].binlog_file'))` + --let $checkpoint_binlog_position = `SELECT JSON_UNQUOTE(JSON_EXTRACT(@checkpoints, '\$[$checkpoint_json_index].position'))` + --let $checkpoint_label = `SELECT JSON_UNQUOTE(JSON_EXTRACT(@checkpoints, '\$[$checkpoint_json_index].test_label'))` + --source ../include/rs_compare_dumps.inc + --inc $checkpoint_json_index +} + +--echo +--echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for +--echo *** the process to terminate. +--source ../include/terminate_proc.inc + +--echo +--echo *** Dropping the table. +DROP TABLE tbl; +--disable_query_log +SET @checkpoints = NULL; +--enable_query_log + +--source ../include/tear_down_binsrv_environment.inc + +# As the Binlog Server Utility interrupts the connection upon timeout, here we +# need to close it on the MySQL server side as well in order to make sure that +# MTR 'check-test' before and after the test produces the same output. +--source ../include/kill_binlog_dump_connection.inc + +--source include/wait_until_count_sessions.inc + +# End of test.