Dirty data encountered during session data storage, resulting in session consistency failure

[jsRuner]

Search before asking

• I had searched in the issues and found no similar issues.

Operating system information

Linux

Python version information

=3.11

DB-GPT version

main

Related scenes

• Chat Data
• Chat Excel
• Chat DB
• Chat Knowledge
• Model Management
• Dashboard
• Plugins

Installation Information

• Installation From Source

• Docker Installation

• Docker Compose Installation

• Cluster Installation

• AutoDL Image

• Other

Device information

no info

Models information

no info

What happened

dbgpt/core/interface/message.py

messages_to_save = message_list[self._has_stored_message_index + 1 :]
        self._has_stored_message_index = len(message_list) - 1
        if self.save_message_independent:
            # Save messages independently
            self.message_storage.save_list(messages_to_save) 
        self.conv_storage.save_or_update(self)

I analyzed the code and found that two tables were operated on in sequence, but no transactions were used. Once an exception occurs, it triggers a session exception. But the probability of this problem occurring may be low, and I'm not sure if it needs to be fixed, but I did encounter it. I cannot determine how it was triggered, but the historical trace is that the chat_history_message table corresponds to 14 messages, and the message_ids of the chat_history table correspond to 12 message ID strings. At this point, no matter how I chat, the front-end will not respond. Because backend insertion of messages will consistently report errors.
The error log is

File "/app/dbgpt/util/executor_utils.py", line 67, in blocking_func_to_async
    return await loop.run_in_executor(executor, run_with_context)
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/app/dbgpt/util/executor_utils.py", line 64, in run_with_context
    return ctx.run(partial(func, *args, **kwargs))
  File "/app/dbgpt/core/interface/message.py", line 1010, in end_current_round
    self.save_to_storage()
  File "/app/dbgpt/core/interface/message.py", line 1029, in save_to_storage
    self.message_storage.save_list(messages_to_save)
  File "/app/dbgpt/core/interface/storage.py", line 265, in save_list
    self.save(d)
  File "/app/dbgpt/storage/metadata/db_storage.py", line 57, in save
    with self.session() as session:
  File "/usr/lib/python3.10/contextlib.py", line 142, in __exit__
    next(self.gen)
  File "/app/dbgpt/storage/metadata/db_storage.py", line 52, in session
    with self.db_manager.session() as session:
  File "/usr/lib/python3.10/contextlib.py", line 142, in __exit__
    next(self.gen)
  File "/app/dbgpt/storage/metadata/db_manager.py", line 249, in session
    session.commit()
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 1972, in commit
    trans.commit(_to_root=True)
  File "<string>", line 2, in commit
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 1257, in commit
    self._prepare_impl()
  File "<string>", line 2, in _prepare_impl
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 1232, in _prepare_impl
    self.session.flush()
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 4296, in flush
    self._flush(objects)
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 4431, in _flush
    with util.safe_reraise():
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/session.py", line 4392, in _flush
    flush_context.execute()
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/unitofwork.py", line 466, in execute
    rec.execute(self)
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/unitofwork.py", line 642, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/persistence.py", line 93, in save_obj
    _emit_insert_statements(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/orm/persistence.py", line 1233, in _emit_insert_statements
    result = connection.execute(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 1421, in execute
    return meth(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/sql/elements.py", line 514, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 1643, in _execute_clauseelement
    ret = self._execute_context(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 1849, in _execute_context
    return self._exec_single_context(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 1989, in _exec_single_context
    self._handle_dbapi_exception(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 2356, in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/base.py", line 1970, in _exec_single_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.10/dist-packages/sqlalchemy/engine/default.py", line 924, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.10/dist-packages/pymysql/cursors.py", line 153, in execute
    result = self._query(query)
  File "/usr/local/lib/python3.10/dist-packages/pymysql/cursors.py", line 322, in _query
    conn.query(q)
  File "/usr/local/lib/python3.10/dist-packages/pymysql/connections.py", line 558, in query
    self._affected_rows = self._read_query_result(unbuffered=unbuffered)
  File "/usr/local/lib/python3.10/dist-packages/pymysql/connections.py", line 822, in _read_query_result
    result.read()
  File "/usr/local/lib/python3.10/dist-packages/pymysql/connections.py", line 1200, in read
    first_packet = self.connection._read_packet()
  File "/usr/local/lib/python3.10/dist-packages/pymysql/connections.py", line 772, in _read_packet
    packet.raise_for_error()
  File "/usr/local/lib/python3.10/dist-packages/pymysql/protocol.py", line 221, in raise_for_error
    err.raise_mysql_exception(self._data)
  File "/usr/local/lib/python3.10/dist-packages/pymysql/err.py", line 143, in raise_mysql_exception
    raise errorclass(errno, errval)
sqlalchemy.exc.IntegrityError: (pymysql.err.IntegrityError) (1062, "Duplicate entry '3d27797a-f701-11ee-a6fa-0242ac13001c-12' for key 'chat_history_message.message_uid_index'")
[SQL: INSERT INTO chat_history_message (conv_uid, `index`, round_index, message_detail, gmt_created, gmt_modified) VALUES (%(conv_uid)s, %(index)s, %(round_index)s, %(message_detail)s, %(gmt_created)s, %(gmt_modified)s)]
[parameters: {'conv_uid': '3d27797a-f701-11ee-a6fa-0242ac13001c', 'index': 12, 'round_index': 5, 'message_detail': '{"type": "human", "data": {"content": "22", "index": 12, "round_index": 5, "additional_kwargs": {"param_type": "DB Select", "param_value": "gold_igpt", "model_name": "spark_proxyllm_1.0.2"}, "example": false}, "index": 12, "round_index": 5}', 'gmt_created': datetime.datetime(2024, 5, 6, 6, 58, 8, 884353), 'gmt_modified': datetime.datetime(2024, 5, 6, 6, 58, 8, 884359)}]

What you expected to happen

Avoid dirty data and ensure session availability

How to reproduce

no info

Additional context

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!