PBS-3: Introduce new <replication_source> section into the main configuration file

CMakeLists.txt

@@ -600,6 +600,10 @@ set(binsrv_source_files
   src/binsrv/replication_mode_type_fwd.hpp
   src/binsrv/replication_mode_type.hpp
 
+  src/binsrv/replication_source_config_fwd.hpp
+  src/binsrv/replication_source_config.hpp
+  src/binsrv/replication_source_config.cpp
+
   src/binsrv/rewrite_config_fwd.hpp
   src/binsrv/rewrite_config.hpp
   src/binsrv/rewrite_config.cpp

README.md

@@ -514,6 +514,11 @@ The Percona Binary Log Server configuration file has the following format.
       "file_size": "128M"
     }
   },
+  "replication_source": {
+    "port": 3307,
+    "read_timeout": 60,
+    "write_timeout": 60
+  },
   "keyring": {
     "uri": "file:///var/lib/pbs/keyring/keyring_data.json"
   },
@@ -584,6 +589,12 @@ If this section is present, then the utility will not split binlog events the sa
 - `<replication.rewrite.base_file_name>` - the base name of the generated binlog file names in the "rewrite" mode. E.g. `rewritten_binlog` will cause `rewritten_binlog.000001`, `rewritten_binlog.000002`, etc. file names to be generated.
 - `<replication.rewrite.file_size>` - the maximum individual binlog file size after reaching which the utility will switch to a new one. The value is expected to be a string containing an integer followed by an optional suffix 'K' / 'M' / 'G' / 'T' / 'P', e.g. /\d+\[KMGTP\]?/. The minimal allowed value of this parameter is `1024` bytes.
 
+#### \<replication_source\> section
+This section configures the built-in MySQL-compatible listener the utility exposes in `pull` mode so that downstream replicas can dump binary log events from it (the utility acts as a replication source).
+- `<replication_source.port>` - the TCP port on which the utility listens for incoming replica connections.
+- `<replication_source.read_timeout>` - the number of seconds the utility will wait to read data from a connected replica before treating the connection as timed out.
+- `<replication_source.write_timeout>` - the number of seconds the utility will wait to write data to a connected replica before treating the connection as timed out.
+
 #### \<keyring\> section
 If this an optional section that specifies keyring configuration parameters. It must be present if the storage has at least one encrypted binlog file.
 - `<keyring.uri>` - specifies location of the keyring JSON data file (currently only 'file://' scheme is supported meaning that the file should be taken from the local file sytem from the path specified in this URI, e.g. `file:///var/lib/pbs/keyring/keyring_data.json`).

main_config.json

@@ -36,6 +36,11 @@
       "file_size": "128M"
     }
   },
+  "replication_source": {
+    "port": 3307,
+    "read_timeout": 60,
+    "write_timeout": 60
+  },
   "keyring": {
     "uri": "file:///home/user/keyring/keyring/keyring_data.json"
   },

mtr/binlog_streaming/include/generate_binsrv_config.inc

@@ -1,6 +1,11 @@
 #
 # Creates a JSON configuration file for the Binlog Server Utility
 #
+# Sets (as an output for the caller):
+#   $binsrv_replication_source_port - TCP port on which the utility
+#     listens for downstream replicas (@@global.port + 1000, unique per
+#     MTR worker).
+#
 # Usage:
 #   --let $binsrv_log_level = trace | debug | info | warning | error | fatal (optional, default: trace)
 #   --let $binsrv_connection_user = repl_user (optional)
@@ -88,6 +93,20 @@ if ($binsrv_connection_host == "")
   }
 }
 
+# The Binlog Server exposes its own MySQL-compatible listener in 'pull'
+# mode. To let multiple MTR workers run pull-mode tests in parallel, each
+# worker gets its own listening port derived from the worker's MySQL port
+# (worker port + 1000). The +1000 offset intentionally jumps well beyond
+# MTR's per-worker port allocation (empirically MTR reserves ports within
+# ~200 of each worker's @@port on GitHub Actions runners), so the derived
+# listener port never collides with a neighbour worker's mysqld/mysqlx or
+# other MTR-reserved ports. An earlier +100 offset produced bind failures
+# ("Address already in use") on CI when the derived port fell inside a
+# sibling worker's range. We expose the computed value via
+# $binsrv_replication_source_port so the test can reuse it (e.g. when
+# connecting a probe client to the listener).
+--let $binsrv_replication_source_port = `SELECT @@global.port + 1000`
+
 eval SET @binsrv_config_json = JSON_OBJECT(
   'logger', JSON_OBJECT(
     'level', '$binsrv_log_level',
@@ -108,6 +127,11 @@ eval SET @binsrv_config_json = JSON_OBJECT(
      'verify_checksum', $binsrv_verify_checksum,
      'mode', '$binsrv_replication_mode'
   ),
+  'replication_source', JSON_OBJECT(
+     'port', $binsrv_replication_source_port,
+     'read_timeout', 60,
+     'write_timeout', 60
+  ),
   'storage', JSON_OBJECT(
      'backend', '$storage_backend',
      'uri', @storage_uri

mtr/binlog_streaming/r/auth_method_switch.result

@@ -17,11 +17,11 @@
 include/read_file_to_var.inc
 
 *** Waiting for the Binlog Server listener to come up on
-*** 127.0.0.1:3307. We probe with bash's /dev/tcp instead of the
-*** mysql client because bash is not ASAN-instrumented and
-*** /dev/tcp uses a plain connect(2), so each attempt is cheap and
-*** measures exactly the "listening on the port" state we care
-*** about.
+*** 127.0.0.1:<BINSRV_LISTENER_PORT>. We probe with bash's /dev/tcp
+*** instead of the mysql client because bash is not
+*** ASAN-instrumented and /dev/tcp uses a plain connect(2), so each
+*** attempt is cheap and measures exactly the "listening on the
+*** port" state we care about.
 
 *** Control: client picks the same plugin the server advertises
 *** (caching_sha2_password). No AuthMethodSwitch is expected on the

mtr/binlog_streaming/t/auth_method_switch.test

@@ -1,7 +1,3 @@
-# The Binlog Server listens on a hard-coded TCP port, so this test can not
-# run in parallel with other tests using the same port.
---source include/not_parallel.inc
-
 --source ../include/have_binsrv.inc
 
 --source ../include/v80_v84_compatibility_defines.inc
@@ -61,11 +57,11 @@ EOF
 
 --echo
 --echo *** Waiting for the Binlog Server listener to come up on
---echo *** 127.0.0.1:3307. We probe with bash's /dev/tcp instead of the
---echo *** mysql client because bash is not ASAN-instrumented and
---echo *** /dev/tcp uses a plain connect(2), so each attempt is cheap and
---echo *** measures exactly the "listening on the port" state we care
---echo *** about.
+--echo *** 127.0.0.1:<BINSRV_LISTENER_PORT>. We probe with bash's /dev/tcp
+--echo *** instead of the mysql client because bash is not
+--echo *** ASAN-instrumented and /dev/tcp uses a plain connect(2), so each
+--echo *** attempt is cheap and measures exactly the "listening on the
+--echo *** port" state we care about.
 --let $max_wait = 300
 --let $iteration = 0
 --let $port_open = 0
@@ -74,7 +70,7 @@ while ($iteration < $max_wait)
   if (!$port_open)
   {
     --error 0, 1
-    --exec bash -c "echo > /dev/tcp/127.0.0.1/3307" 2>/dev/null
+    --exec bash -c "echo > /dev/tcp/127.0.0.1/$binsrv_replication_source_port" 2>/dev/null
     --let $port_status = $__error
     if ($port_status == 0)
     {
@@ -90,15 +86,15 @@ while ($iteration < $max_wait)
 }
 if (!$port_open)
 {
-  --die The Binlog Server listener did not become reachable on 3307 within 300 seconds
+  --die The Binlog Server listener did not become reachable on its port within 300 seconds
 }
 
 --echo
 --echo *** Control: client picks the same plugin the server advertises
 --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=3307 --user=rpl --password=password --default-auth=caching_sha2_password --skip-column-names -e "$probe_query" >/dev/null 2>&1
+--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
 
 --echo
 --echo *** Trigger: client forces mysql_native_password in its handshake so
@@ -109,7 +105,7 @@ if (!$port_open)
 --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=3307 --user=rpl --password=password --default-auth=mysql_native_password --skip-column-names -e "$probe_query" >/dev/null 2>&1
+--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
 
 --echo
 --echo *** Sending SIGTERM to the Binlog Server Utility and waiting for the

mtr/binlog_streaming/t/binlog_flush.test

@@ -1,7 +1,3 @@
-# temporarily marking this test as non-parallel because of the hardcoded
-# listening port
---source include/not_parallel.inc
-
 # The purpose of the test is to validate that PBS doesn't flush its internal binlog buffer
 # to the file every time the whole transaction is replicated.
 

mtr/binlog_streaming/t/pull_mode.test

@@ -1,7 +1,3 @@
-# temporarily marking this test as non-parallel because of the hardcoded
-# listening port
---source include/not_parallel.inc
-
 --source ../include/have_binsrv.inc
 
 --source ../include/v80_v84_compatibility_defines.inc
@@ -101,7 +97,15 @@ FLUSH BINARY LOGS;
 --echo *** binary log.
 # We grep the Binlog Server Utility log file in a loop until we encounter the
 # fourth binary log file name.
---let $max_number_of_attempts = 60
+# The wait budget has to cover several reconnect cycles: the outbound
+# read_timeout is 3s and the idle wait between reconnects is 1s (see the
+# $binsrv_read_timeout / $binsrv_idle_time above), so binsrv needs
+# multiple ~5s round trips to walk through the three preceding binlog
+# files. Under sanitized (ASAN) builds and parallel MTR execution those
+# round trips slow down further, so 60 iterations is not enough - 300
+# gives it several minutes and still bails out well before the outer
+# testcase timeout kicks in.
+--let $max_number_of_attempts = 300
 --let $iteration = 0
 while($iteration < $max_number_of_attempts)
 {

src/binsrv/main_config.cpp

@@ -60,6 +60,7 @@ void main_config::validate() const {
   root().get<"connection">().validate();
   root().get<"storage">().validate();
   root().get<"replication">().validate();
+  root().get<"replication_source">().validate();
 }
 
 } // namespace binsrv

src/binsrv/main_config.hpp

@@ -18,10 +18,11 @@
 
 #include "binsrv/main_config_fwd.hpp" // IWYU pragma: export
 
-#include "binsrv/keyring_config.hpp"     // IWYU pragma: export
-#include "binsrv/logger_config.hpp"      // IWYU pragma: export
-#include "binsrv/replication_config.hpp" // IWYU pragma: export
-#include "binsrv/storage_config.hpp"     // IWYU pragma: export
+#include "binsrv/keyring_config.hpp"            // IWYU pragma: export
+#include "binsrv/logger_config.hpp"             // IWYU pragma: export
+#include "binsrv/replication_config.hpp"        // IWYU pragma: export
+#include "binsrv/replication_source_config.hpp" // IWYU pragma: export
+#include "binsrv/storage_config.hpp"            // IWYU pragma: export
 
 #include "easymysql/connection_config.hpp" // IWYU pragma: export
 
@@ -33,11 +34,12 @@ class [[nodiscard]] main_config {
 private:
   using impl_type = util::nv_tuple<
       // clang-format off
-      util::nv<"logger"     , logger_config>,
-      util::nv<"connection" , easymysql::connection_config>,
-      util::nv<"replication", binsrv::replication_config>,
-      util::nv<"keyring"    , optional_keyring_config>,
-      util::nv<"storage"    , storage_config>
+      util::nv<"logger"            , logger_config>,
+      util::nv<"connection"        , easymysql::connection_config>,
+      util::nv<"replication"       , binsrv::replication_config>,
+      util::nv<"replication_source", binsrv::replication_source_config>,
+      util::nv<"keyring"           , optional_keyring_config>,
+      util::nv<"storage"           , storage_config>
       // clang-format on
       >;
 

src/binsrv/replication_source_config.cpp

@@ -0,0 +1,42 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#include "binsrv/replication_source_config.hpp"
+
+#include <stdexcept>
+
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+void replication_source_config::validate() const {
+  if (get<"port">() == 0U) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source config: "
+        "port must be greater than 0");
+  }
+  if (get<"read_timeout">() == 0U) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source config: "
+        "read_timeout must be greater than 0");
+  }
+  if (get<"write_timeout">() == 0U) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source config: "
+        "write_timeout must be greater than 0");
+  }
+}
+
+} // namespace binsrv

src/binsrv/replication_source_config.hpp

@@ -0,0 +1,40 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#ifndef BINSRV_REPLICATION_SOURCE_CONFIG_HPP
+#define BINSRV_REPLICATION_SOURCE_CONFIG_HPP
+
+#include "binsrv/replication_source_config_fwd.hpp" // IWYU pragma: export
+
+#include <cstdint>
+
+#include "util/nv_tuple.hpp"
+
+namespace binsrv {
+
+struct [[nodiscard]] replication_source_config
+    : util::nv_tuple<
+          // clang-format off
+          util::nv<"port"         , std::uint16_t>,
+          util::nv<"read_timeout" , std::uint32_t>,
+          util::nv<"write_timeout", std::uint32_t>
+          // clang-format on
+          > {
+  void validate() const;
+};
+
+} // namespace binsrv
+
+#endif // BINSRV_REPLICATION_SOURCE_CONFIG_HPP

src/binsrv/replication_source_config_fwd.hpp

@@ -0,0 +1,25 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#ifndef BINSRV_REPLICATION_SOURCE_CONFIG_FWD_HPP
+#define BINSRV_REPLICATION_SOURCE_CONFIG_FWD_HPP
+
+namespace binsrv {
+
+struct replication_source_config;
+
+} // namespace binsrv
+
+#endif // BINSRV_REPLICATION_SOURCE_CONFIG_FWD_HPP