Hi, I'm encountering difficulties with getting Meilisync to function properly due to a recurring issue of lost connection, accompanied by an error upon attempting to reconnect. I'm utilizing MySQL as the data source for Meilisync.
Here's what I've observed:
The ctl_conn of the binlogstream is consistently lost after a short period.
Whenever a database change is made, an exception is raised: asyncmy.errors.OperationalError: (2013, 'Lost connection to MySQL server during query').
Despite attempts to recreate the binlog stream upon encountering this exception, it seems that the self.ctl_conn.connect() line fails to execute as expected.
Subsequently, an additional exception is triggered: AttributeError: 'NoneType' object has no attribute 'write', indicating that the self._writer attribute of the connection remains None even after connect() is invoked, which should not be the case.
I managed to make a workound by moving self.ctl_conn initialization inside the _create_stream function on the mysql source class (i would happily make a PR with this change if necessary):
async def _create_stream(self):
self.ctl_conn = await asyncmy.connect(**self.kwargs)
await self.ctl_conn.connect()
self.stream = BinLogStream(
self.conn,
self.ctl_conn,
server_id=self.server_id,
master_log_file=self.progress["master_log_file"],
master_log_position=int(self.progress["master_log_position"]),
resume_stream=True,
blocking=True,
only_schemas=[self.database],
only_tables=[f"{self.database}.{table}" for table in self.tables],
only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent],
)
async def __aiter__(self):
self.conn = await asyncmy.connect(**self.kwargs)
if not self.progress:
self.progress = await self.get_current_progress()
yield ProgressEvent(
progress=self.progress,
)
await self._create_stream()
while True:
try:
async for event in self.stream:
..:
except OperationalError as e:
logger.exception(f"Binlog stream error: {e}, sleep 10s and retry...")
await asyncio.sleep(10)
try:
await self.stream.close()
self.ctl_conn.close()
await self._create_stream()
except Exception as e:
logger.exception(f"Recreate binlog stream error: {e}")
This adjustment resolves the reconnection issue, yet I aim to discover a method to maintain the connection alive to prevent recurrent OperationalError exception.
Here's the log of the OperationalError
2024-02-06 20:15:59.072 | ERROR | meilisync.source.mysql:__aiter__:130 - Binlog stream error: (2013, 'Lost connection to MySQL server during query'), sleep 10s and retry...
Traceback (most recent call last):
File "/usr/local/bin/meilisync", line 6, in <module>
sys.exit(app())
│ │ └ <typer.main.Typer object at 0x7f991300ba70>
│ └ <built-in function exit>
└ <module 'sys' (built-in)>
File "/usr/local/lib/python3.12/site-packages/typer/main.py", line 311, in __call__
return get_command(self)(*args, **kwargs)
│ │ │ └ {}
│ │ └ ()
│ └ <typer.main.Typer object at 0x7f991300ba70>
└ <function get_command at 0x7f9911e26340>
File "/usr/local/lib/python3.12/site-packages/click/core.py", line 1157, in __call__
return self.main(*args, **kwargs)
│ │ │ └ {}
│ │ └ ()
│ └ <function TyperGroup.main at 0x7f9911e24c20>
└ <TyperGroup callback>
File "/usr/local/lib/python3.12/site-packages/typer/core.py", line 778, in main
return _main(
└ <function _main at 0x7f9911e1fb00>
File "/usr/local/lib/python3.12/site-packages/typer/core.py", line 216, in _main
rv = self.invoke(ctx)
│ │ └ <click.core.Context object at 0x7f990edf7c20>
│ └ <function MultiCommand.invoke at 0x7f99122447c0>
└ <TyperGroup callback>
File "/usr/local/lib/python3.12/site-packages/click/core.py", line 1688, in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
│ │ │ │ └ <click.core.Context object at 0x7f990fe15040>
│ │ │ └ <function Command.invoke at 0x7f9912244180>
│ │ └ <TyperCommand start>
│ └ <click.core.Context object at 0x7f990fe15040>
└ <function MultiCommand.invoke.<locals>._process_result at 0x7f990e544040>
File "/usr/local/lib/python3.12/site-packages/click/core.py", line 1434, in invoke
return ctx.invoke(self.callback, **ctx.params)
│ │ │ │ │ └ {}
│ │ │ │ └ <click.core.Context object at 0x7f990fe15040>
│ │ │ └ <function start at 0x7f990e53a020>
│ │ └ <TyperCommand start>
│ └ <function Context.invoke at 0x7f9912222ac0>
└ <click.core.Context object at 0x7f990fe15040>
File "/usr/local/lib/python3.12/site-packages/click/core.py", line 783, in invoke
return __callback(*args, **kwargs)
│ └ {}
└ ()
File "/usr/local/lib/python3.12/site-packages/typer/main.py", line 683, in wrapper
return callback(**use_params) # type: ignore
│ └ {'context': <click.core.Context object at 0x7f990fe15040>}
└ <function start at 0x7f990e539e40>
File "/meilisync/meilisync/main.py", line 155, in start
asyncio.run(run())
│ │ └ <function start.<locals>.run at 0x7f990e51d3a0>
│ └ <function run at 0x7f9912dc5e40>
└ <module 'asyncio' from '/usr/local/lib/python3.12/asyncio/__init__.py'>
File "/usr/local/lib/python3.12/asyncio/runners.py", line 194, in run
return runner.run(main)
│ │ └ <coroutine object start.<locals>.run at 0x7f990e512ea0>
│ └ <function Runner.run at 0x7f991237bce0>
└ <asyncio.runners.Runner object at 0x7f990e51a300>
File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
│ │ │ └ <Task pending name='Task-4' coro=<start.<locals>.run() running at /meilisync/meilisync/main.py:153> wait_for=<_GatheringFutur...
│ │ └ <function BaseEventLoop.run_until_complete at 0x7f9912379940>
│ └ <_UnixSelectorEventLoop running=True closed=False debug=False>
└ <asyncio.runners.Runner object at 0x7f990e51a300>
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 671, in run_until_complete
self.run_forever()
│ └ <function BaseEventLoop.run_forever at 0x7f99123798a0>
└ <_UnixSelectorEventLoop running=True closed=False debug=False>
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 638, in run_forever
self._run_once()
│ └ <function BaseEventLoop._run_once at 0x7f991237b6a0>
└ <_UnixSelectorEventLoop running=True closed=False debug=False>
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 1971, in _run_once
handle._run()
│ └ <function Handle._run at 0x7f99124d9b20>
└ <Handle Task.task_wakeup(<Future finished result=None>)>
File "/usr/local/lib/python3.12/asyncio/events.py", line 84, in _run
self._context.run(self._callback, *self._args)
│ │ │ │ │ └ <member '_args' of 'Handle' objects>
│ │ │ │ └ <Handle Task.task_wakeup(<Future finished result=None>)>
│ │ │ └ <member '_callback' of 'Handle' objects>
│ │ └ <Handle Task.task_wakeup(<Future finished result=None>)>
│ └ <member '_context' of 'Handle' objects>
└ <Handle Task.task_wakeup(<Future finished result=None>)>
File "/meilisync/meilisync/main.py", line 102, in _
async for event in source:
│ └ <meilisync.source.mysql.MySQL object at 0x7f990e9523c0>
└ Event(progress={'master_log_file': 'mysql-bin-changelog.492584', 'master_log_position': 3540744}, type=<EventType.update: 'up...
> File "/meilisync/meilisync/source/mysql.py", line 108, in __aiter__
async for event in self.stream:
│ │ └ <asyncmy.replication.binlogstream.BinLogStream object at 0x7f990e52e5a0>
│ └ <meilisync.source.mysql.MySQL object at 0x7f990e9523c0>
└ <asyncmy.replication.row_events.UpdateRowsEvent object at 0x7f990e56c110>
File "/usr/local/lib/python3.12/site-packages/asyncmy/replication/binlogstream.py", line 374, in __anext__
ret = await self._read()
│ └ <function BinLogStream._read at 0x7f990fe5eac0>
└ <asyncmy.replication.binlogstream.BinLogStream object at 0x7f990e52e5a0>
File "/usr/local/lib/python3.12/site-packages/asyncmy/replication/binlogstream.py", line 310, in _read
await binlog_event.init()
│ └ <function BinLogPacket.init at 0x7f990fe0bb00>
└ <asyncmy.replication.packets.BinLogPacket object at 0x7f991021eff0>
File "/usr/local/lib/python3.12/site-packages/asyncmy/replication/packets.py", line 140, in init
self.event and await self.event.init()
│ │ │ │ └ <function TableMapEvent.init at 0x7f990fe5d260>
│ │ │ └ <asyncmy.replication.row_events.TableMapEvent object at 0x7f990e56d6d0>
│ │ └ <asyncmy.replication.packets.BinLogPacket object at 0x7f991021eff0>
│ └ <asyncmy.replication.row_events.TableMapEvent object at 0x7f990e56d6d0>
└ <asyncmy.replication.packets.BinLogPacket object at 0x7f991021eff0>
File "/usr/local/lib/python3.12/site-packages/asyncmy/replication/row_events.py", line 580, in init
await self._connection._get_table_information(self.schema, self.table_name)
│ │ │ │ │ │ └ 'test_courses'
│ │ │ │ │ └ <asyncmy.replication.row_events.TableMapEvent object at 0x7f990e56d6d0>
│ │ │ │ └ 'test_schema'
│ │ │ └ <asyncmy.replication.row_events.TableMapEvent object at 0x7f990e56d6d0>
│ │ └ <bound method BinLogStream._get_table_information of <asyncmy.replication.binlogstream.BinLogStream object at 0x7f990e52e5a0>>
│ └ <asyncmy.connection.Connection object at 0x7f990e52f8c0>
└ <asyncmy.replication.row_events.TableMapEvent object at 0x7f990e56d6d0>
File "/usr/local/lib/python3.12/site-packages/asyncmy/replication/binlogstream.py", line 350, in _get_table_information
await cursor.execute(
│ └ <cyfunction Cursor.execute at 0x7f990fe07440>
└ <asyncmy.cursors.DictCursor object at 0x7f990e97d760>
File "asyncmy/cursors.pyx", line 179, in execute
result = await self._query(query)
File "asyncmy/cursors.pyx", line 364, in _query
await conn.query(q)
File "asyncmy/connection.pyx", line 494, in query
await self._read_query_result(unbuffered=unbuffered)
File "asyncmy/connection.pyx", line 682, in _read_query_result
await result.read()
File "asyncmy/connection.pyx", line 1069, in read
first_packet = await self.connection.read_packet()
File "asyncmy/connection.pyx", line 623, in read_packet
raise errors.OperationalError(
│ └ <class 'asyncmy.errors.OperationalError'>
└ <module 'asyncmy.errors' from '/usr/local/lib/python3.12/site-packages/asyncmy/errors.cpython-312-x86_64-linux-gnu.so'>
asyncmy.errors.OperationalError: (2013, 'Lost connection to MySQL server during query')
Hi, I'm encountering difficulties with getting Meilisync to function properly due to a recurring issue of lost connection, accompanied by an error upon attempting to reconnect. I'm utilizing MySQL as the data source for Meilisync.
Here's what I've observed:
asyncmy.errors.OperationalError: (2013, 'Lost connection to MySQL server during query').self.ctl_conn.connect()line fails to execute as expected.I managed to make a workound by moving
self.ctl_conninitialization inside the_create_streamfunction on the mysql source class (i would happily make a PR with this change if necessary):This adjustment resolves the reconnection issue, yet I aim to discover a method to maintain the connection alive to prevent recurrent
OperationalErrorexception.Here's the log of the
OperationalErrorHere's the
AttributeErrorException