Reading binlog failed, When setting rpl_semi_sync_slave=1

[liaobaikai]

The master is configured with semisync mode
[root@localhost][(none)]> select version();
+-----------+
| version() |
+-----------+
| 8.0.41 |
+-----------+
1 row in set (0.00 sec)

[root@localhost][(none)]> SHOW variables LIKE 'rpl_semi_sync_%';
+-------------------------------------------+------------+
| Variable_name | Value |
+-------------------------------------------+------------+
| rpl_semi_sync_master_enabled | ON |
| rpl_semi_sync_master_timeout | 10000 |
| rpl_semi_sync_master_trace_level | 32 |
| rpl_semi_sync_master_wait_for_slave_count | 1 |
| rpl_semi_sync_master_wait_no_slave | ON |
| rpl_semi_sync_master_wait_point | AFTER_SYNC |
+-------------------------------------------+------------+
6 rows in set (0.00 sec)

let opts = OptsBuilder::default()
        .ip_or_hostname("localhost")
        .user(Some("root"))
        .pass(Some("123456"))
        .db_name(Some("test"))
        .init(vec!["SET @rpl_semi_sync_replica = 1, @rpl_semi_sync_slave = 1"]);

    let pool = Pool::new(opts);
    let mut conn = pool.get_conn().await?;
    println!("conn: {:?}", conn);

    //conn.query_drop("SET @rpl_semi_sync_replica = 1, @rpl_semi_sync_slave = 1").await?;

    let request = BinlogStreamRequest::new(3000)
        .with_filename(b"mysql-bin.000001")
        .with_pos(157);
    let mut binlog_stream = conn.get_binlog_stream(request).await?;

    let mut events_num = 0;

    while let Some(event) = binlog_stream.next().await {
        let event = event.unwrap();
        events_num += 1;

        // assert that event type is known
        event.header().event_type().unwrap();

        println!("event: {:?}", event);

        // iterate over rows of an event
        if let EventData::RowsEvent(re) = event.read_data()?.unwrap() {
            let tme = binlog_stream.get_tme(re.table_id());
            for row in re.rows(tme.unwrap()) {
                row.unwrap();
            }
        }
    }

thread 'main' panicked at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mysql_common-0.35.5/src/binlog/events/mod.rs:126:37:
called Result::unwrap() on an Err value: Error { kind: UnexpectedEof, message: "failed to fill whole buffer" }
stack backtrace:
0: __rustc::rust_begin_unwind
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library/std/src/panicking.rs:697:5
1: core::panicking::panic_fmt
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library/core/src/panicking.rs:75:14
2: core::result::unwrap_failed
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library/core/src/result.rs:1732:5
3: core::result::Result<T,E>::unwrap
at /Users/lbk/.rustup/toolchains/stable-x86_64-apple-darwin/lib/rustlib/src/rust/library/core/src/result.rs:1137:23
4: mysql_common::binlog::events::Event::read
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mysql_common-0.35.5/src/binlog/events/mod.rs:126:9
5: mysql_common::binlog::EventStreamReader::read
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mysql_common-0.35.5/src/binlog/mod.rs:160:21
6: <mysql_async::conn::binlog_stream::BinlogStream as futures_core::stream::Stream>::poll_next
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mysql_async-0.36.1/src/conn/binlog_stream/mod.rs:165:19
7: futures_util::stream::stream::StreamExt::poll_next_unpin
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/futures-util-0.3.31/src/stream/stream/mod.rs:1638:9
8: <futures_util::stream::stream::next::Next as core::future::future::Future>::poll
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/futures-util-0.3.31/src/stream/stream/next.rs:32:9
9: mysqltest::main::{{closure}}
at ./src/main.rs:39:50
10: tokio::runtime::park::CachedParkThread::block_on::{{closure}}
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/park.rs:285:60
11: tokio::task::coop::with_budget
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/task/coop/mod.rs:167:5
12: tokio::task::coop::budget
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/task/coop/mod.rs:133:5
13: tokio::runtime::park::CachedParkThread::block_on
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/park.rs:285:31
14: tokio::runtime::context::blocking::BlockingRegionGuard::block_on
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/context/blocking.rs:66:9
15: tokio::runtime::scheduler::multi_thread::MultiThread::block_on::{{closure}}
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/scheduler/multi_thread/mod.rs:87:13
16: tokio::runtime::context::runtime::enter_runtime
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/context/runtime.rs:65:16
17: tokio::runtime::scheduler::multi_thread::MultiThread::block_on
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/scheduler/multi_thread/mod.rs:86:9
18: tokio::runtime::runtime::Runtime::block_on_inner
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/runtime.rs:358:45
19: tokio::runtime::runtime::Runtime::block_on
at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.47.1/src/runtime/runtime.rs:330:13
20: mysqltest::main
at ./src/main.rs:64:5
21: core::ops::function::FnOnce::call_once
at /Users/lbk/.rustup/toolchains/stable-x86_64-apple-darwin/lib/rustlib/src/rust/library/core/src/ops/function.rs:250:5
note: Some details are omitted, run with RUST_BACKTRACE=full for a verbose backtrace.

[blackbeam]

@liaobaikai, hi

The should_read_binlog test have passed successfully on MySql 8.4 with the following settings.
The SET @rpl_semi_sync_replica = 1, @rpl_semi_sync_slave = 1 query was executed at runtime.

mysql> SHOW variables LIKE 'rpl_semi_sync_%';
+---------------------------------------------+------------+
| Variable_name                               | Value      |
+---------------------------------------------+------------+
| rpl_semi_sync_replica_enabled               | ON         |
| rpl_semi_sync_replica_trace_level           | 32         |
| rpl_semi_sync_source_enabled                | ON         |
| rpl_semi_sync_source_timeout                | 10000      |
| rpl_semi_sync_source_trace_level            | 32         |
| rpl_semi_sync_source_wait_for_replica_count | 1          |
| rpl_semi_sync_source_wait_no_replica        | ON         |
| rpl_semi_sync_source_wait_point             | AFTER_SYNC |
+---------------------------------------------+------------+
8 rows in set (0,01 sec)

I don't think the reason is in the rpl_semi_sync_replica or rpl_semi_sync_slave values (I can't google neither of these 🤔)

[liaobaikai]

Thanks,

I still get the same error in the mysql8.4,

Database Settings:

mysql> SHOW variables LIKE 'rpl_semi_sync_%';
+-------------------------------------------+------------+
| Variable_name                             | Value      |
+-------------------------------------------+------------+
| rpl_semi_sync_master_enabled              | ON         |
| rpl_semi_sync_master_timeout              | 10000      |
| rpl_semi_sync_master_trace_level          | 32         |
| rpl_semi_sync_master_wait_for_slave_count | 1          |
| rpl_semi_sync_master_wait_no_slave        | ON         |
| rpl_semi_sync_master_wait_point           | AFTER_SYNC |
| rpl_semi_sync_slave_enabled               | OFF        |
| rpl_semi_sync_slave_trace_level           | 32         |
+-------------------------------------------+------------+
8 rows in set (0.00 sec)

mysql> 
mysql> 
mysql> select version();
+-----------+
| version() |
+-----------+
| 8.4.6     |
+-----------+
1 row in set (0.01 sec)

Rust Examples:

use std::time::Duration;

use futures_util::StreamExt;
use mysql_async::{BinlogStreamRequest, OptsBuilder, Pool, prelude::*};
use mysql_common::binlog::events::EventData;
use tokio::time::timeout;

#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    let opts = OptsBuilder::default()
        .ip_or_hostname("localhost")
        .user(Some("root"))
        .pass(Some("xxxxxxxx"))
        .tcp_port(3306)
        .init(vec!["SET @rpl_semi_sync_replica = 1, @rpl_semi_sync_slave = 1"]);

    let pool = Pool::new(opts);
    let conn = pool.get_conn().await?;
    println!("conn: {:?}", conn);

    let request = BinlogStreamRequest::new(3000)
        .with_filename(b"binlog.000069")
        .with_pos(578433757);
    let mut binlog_stream = conn.get_binlog_stream(request).await?;

    let mut events_num = 0;
    while let Some(event) = binlog_stream.next().await {

        let event = event.unwrap();
        events_num += 1;

        // assert that event type is known
        event.header().event_type().unwrap();

        println!("event: {:?}", event);

        // iterate over rows of an event
        if let EventData::RowsEvent(re) = event.read_data()?.unwrap() {
            let tme = binlog_stream.get_tme(re.table_id());
            for row in re.rows(tme.unwrap()) {
                row.unwrap();
            }
        }
    }

    assert!(events_num > 0);
    timeout(Duration::from_secs(10), binlog_stream.close())
        .await
        .unwrap()
        .unwrap();
    println!("exit");

    Ok(())
}

LogOuput:
lbk@LBKdeMacBook-Pro mysqltest % /Users/lbk/.vscode/extensions/vadimcn.vscode-lldb-1.11.5/adapter/codelldb terminal-agent --connect=52920
conn: Conn { inner: Conn { connection id: 10, server version: (8, 4, 6), pool: Some(Pool { opts: Opts { inner: InnerOpts { mysql_opts: MysqlOpts { user: Some("root"), pass: Some("baikai#1234"), db_name: None, tcp_keepalive: None, tcp_nodelay: true, local_infile_handler: None, pool_opts: PoolOpts { constraints: PoolConstraints { min: 10, max: 100 }, inactive_connection_ttl: 0ns, ttl_check_interval: 30s, abs_conn_ttl: None, abs_conn_ttl_jitter: None, reset_connection: true }, conn_ttl: None, init: ["SET @rpl_semi_sync_slave = 1"], setup: [], stmt_cache_size: 32, ssl_opts: None, prefer_socket: true, socket: None, compression: None, max_allowed_packet: None, wait_timeout: None, secure_auth: true, client_found_rows: false, enable_cleartext_plugin: false }, address: HostPort { host: "localhost", port: 3310, resolved_ips: None } } }, inner: Inner { metrics: Metrics { connection_count: 1, connections_in_pool: 0, active_wait_requests: 0, create_failed: 0, discarded_superfluous_connection: 0, discarded_unestablished_connection: 0, dirty_connection_return: 0, discarded_expired_connection: 0, resetting_connection: 0, discarded_error_during_cleanup: 0, connection_returned_to_pool: 0 }, close: false, closed: false, exchange: Mutex { data: Exchange { waiting: Waitlist { queue: [] }, available: [], exist: 1, recycler: None }, poisoned: false, .. } }, drop: UnboundedSender { chan: Tx { inner: Chan { tx: Tx { block_tail: 0x7fe6ccd04680, tail_position: 0 }, semaphore: Semaphore(0), rx_waker: AtomicWaker, tx_count: 2, rx_fields: "..." } } } }), pending_result: Ok(None), tx_status: None, stream: Some(Stream (endpoint=Socket(Socket { inner: PollEvented { io: Some(UnixStream { fd: FileDesc(OwnedFd { fd: 12 }), local: (unnamed), peer: "/Users/lbk/mysql-8.4.6-macos15-x86_64/data/mysqld.sock\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0" (pathname) }) } }))), options: Opts { inner: InnerOpts { mysql_opts: MysqlOpts { user: Some("root"), pass: Some("baikai#1234"), db_name: None, tcp_keepalive: None, tcp_nodelay: true, local_infile_handler: None, pool_opts: PoolOpts { constraints: PoolConstraints { min: 10, max: 100 }, inactive_connection_ttl: 0ns, ttl_check_interval: 30s, abs_conn_ttl: None, abs_conn_ttl_jitter: None, reset_connection: true }, conn_ttl: None, init: ["SET @rpl_semi_sync_slave = 1"], setup: [], stmt_cache_size: 32, ssl_opts: None, prefer_socket: true, socket: Some("/Users/lbk/mysql-8.4.6-macos15-x86_64/data/mysqld.sock"), compression: None, max_allowed_packet: None, wait_timeout: None, secure_auth: true, client_found_rows: false, enable_cleartext_plugin: false }, address: HostPort { host: "localhost", port: 3310, resolved_ips: None } } }, server_key: None, auth_plugin: CachingSha2Password } }

thread 'main' panicked at /Users/lbk/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/mysql_common-0.35.5/src/binlog/events/mod.rs:126:37:
called Result::unwrap() on an Err value: Error { kind: UnexpectedEof, message: "failed to fill whole buffer" }
note: run with RUST_BACKTRACE=1 environment variable to display a backtrace

PS:
The @rpl_semi_sync_replica or @rpl_semi_sync_slave in MySQL Source Code:
mysql-8.0.43/plugin/semisync/semisync_replica_plugin.cc:: repl_semi_slave_request_dump

/*
    Tell master dump thread that we want to do semi-sync
    replication
  */
  const char *query =
      "SET @rpl_semi_sync_replica = 1, @rpl_semi_sync_slave = 1";
  if (mysql_real_query(mysql, query, static_cast<ulong>(strlen(query)))) {
    LogPluginErr(ERROR_LEVEL, ER_SEMISYNC_REPLICA_SET_FAILED);
    return 1;
  }

Or
MariaDB Source Code:
mariadb-10.3.39/sql/semisync_slave.cc:: request_transmit

/*
   Tell master dump thread that we want to do semi-sync
   replication
  */
  query= "SET @rpl_semi_sync_slave= 1";
  if (mysql_real_query(mysql, query, (ulong)strlen(query)))
  {
    sql_print_error("Set 'rpl_semi_sync_slave=1' on master failed");
    return 1;
  }

[blackbeam]

Since I couldn't reproduce the issue, could you please build your program using a local copy of mysql_async 0.36.1 and dump the contents of the packet here?

[liaobaikai]

Thanks,

My version is also mysql_async 0.36.1, a newly built project yesterday

Cargo.toml

[package]
name = "mysqltest"
version = "0.1.0"
edition = "2024"

[dependencies]
futures-util = "0.3.31"
mysql_async = {version = "0.36.1", features = ["binlog"]}
mysql_common = "0.35.5"
tokio = {version = "1.47.1", features = ["full"]}

mysql:9.3.0
packet: PooledBuf([0, 239, 0, 0, 0, 0, 0, 4, 1, 0, 0, 0, 40, 0, 0, 0, 0, 0, 0, 0, 32, 0, 158, 0, 0, 0, 0, 0, 0, 0, 98, 105, 110, 108, 111, 103, 46, 48, 48, 48, 48, 48, 53], BufferPool { buffer_size_cap: 4194304, buffer_init_cap: 0, pool: ArrayQueue { .. } })

mysql:8.0.39
packet: PooledBuf([0, 239, 0, 0, 0, 0, 0, 4, 1, 0, 0, 0, 40, 0, 0, 0, 0, 0, 0, 0, 32, 0, 157, 0, 0, 0, 0, 0, 0, 0, 98, 105, 110, 108, 111, 103, 46, 48, 48, 48, 48, 48, 50], BufferPool { buffer_size_cap: 4194304, buffer_init_cap: 0, pool: ArrayQueue { .. } })

mariadb:10.3.39
packet: PooledBuf([0, 239, 0, 0, 0, 0, 0, 4, 232, 3, 0, 0, 47, 0, 0, 0, 0, 0, 0, 0, 32, 0, 72, 1, 0, 0, 0, 0, 0, 0, 109, 121, 115, 113, 108, 45, 98, 105, 110, 46, 48, 48, 48, 48, 48, 50, 114, 122, 5, 49], BufferPool { buffer_size_cap: 4194304, buffer_init_cap: 0, pool: ArrayQueue { .. } })

[liaobaikai]

blackbeam, hi
This seems incompatible with read_binlog_stream in semi-sync.

Replication event packet header:
. slave semi-sync off: 1 byte - (0)
. slave semi-sync on: 3 byte - (0, 0xef, 0/1}

[liaobaikai]

I can fix here through this,

if first_byte == Some(0) {
            println!("packet: {:?}", packet);
            let mut event_data = &packet[1..];
            if event_data[0] == 0xef {
                println!("slave semi-sync on");
                if event_data[1] == 1 {
                    println!("slave semi-sync status ON");
                } else {
                    println!("slave semi-sync status OFF");
                }
                event_data = &packet[3..];
            }

            match self.esr.read(event_data) {

[liaobaikai]

It seems that semi synchronization is a challenge, and then the second problem arises , here

thread 'main' panicked at src\main.rs:37:27:
called Result::unwrap() on an Err value: Io(Io(Custom { kind: Other, error: "packet out of order" }))
stack backtrace:
0: __rustc::rust_begin_unwind
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library\std\src/panicking.rs:697:5
1: core::panicking::panic_fmt
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library\core\src/panicking.rs:75:14
2: core::result::unwrap_failed
at /rustc/6b00bc3880198600130e1cf62b8f8a93494488cc/library\core\src/result.rs:1732:5
3: core::result::Result<T,E>::unwrap
at C:\Users\BK-liao.rustup\toolchains\stable-x86_64-pc-windows-gnu\lib/rustlib/src/rust\library\core\src\result.rs:1137:23
4: mysqltest::pull_binlog_events::{{closure}}
at .\src\main.rs:37:21
5: mysqltest::main::{{closure}}
at .\src\main.rs:111:70