From dc740604d295bf8a69aeb374259321f82994d803 Mon Sep 17 00:00:00 2001 From: u7-u7 <116776533+u7-u7@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:19:55 +0800 Subject: [PATCH] fix(mysql): merge minute metrics under one row lock (#97) store_metrics_minute() read the row, merged counters in Python, then wrote with a separate SELECT + UPDATE/INSERT. Two instances flushing the same minute could both read the old value and overwrite each other, and a concurrent insert could lose the whole batch to a duplicate-key error that the outer except swallowed. Run the read-merge-write in one transaction with SELECT ... FOR UPDATE and retry once on a duplicate key or deadlock. Add unit regressions for the insert and merge paths. Co-authored-by: qiye.sjh Co-authored-by: Claude Code --- hy_memory/data/cache_mysql.py | 75 ++++++++++++++----- tests/test_mysql_metrics_atomic.py | 111 +++++++++++++++++++++++++++++ 2 files changed, 170 insertions(+), 16 deletions(-) create mode 100644 tests/test_mysql_metrics_atomic.py diff --git a/hy_memory/data/cache_mysql.py b/hy_memory/data/cache_mysql.py index 110f9f5..31adc72 100644 --- a/hy_memory/data/cache_mysql.py +++ b/hy_memory/data/cache_mysql.py @@ -464,24 +464,67 @@ class MysqlCache(CacheBase): # ================================================================ async def store_metrics_minute(self, minute_ts: str, data: dict) -> None: + import aiomysql + try: now = datetime.now().isoformat() - # 尝试 UPSERT - existing = await self._fetchone( - "SELECT data FROM system_metrics WHERE minute_ts = %s", (minute_ts,) - ) - if existing: - old_data = json.loads(existing["data"]) - merged = self._merge_metric_buckets(old_data, data) - await self._execute( - "UPDATE system_metrics SET data = %s, created_at = %s WHERE minute_ts = %s", - (json.dumps(merged, default=str), now, minute_ts), - ) - else: - await self._execute( - "INSERT INTO system_metrics (minute_ts, data, created_at) VALUES (%s, %s, %s)", - (minute_ts, json.dumps(data, default=str), now), - ) + # Read-merge-write must be one transaction: two instances flushing + # the same minute used to overwrite each other's counters, and a + # concurrent insert could lose the whole batch to a duplicate key. + for attempt in range(2): + try: + async with self._pool.acquire() as conn: + await conn.begin() + try: + async with conn.cursor(aiomysql.DictCursor) as cur: + await cur.execute( + "SELECT data FROM system_metrics " + "WHERE minute_ts = %s FOR UPDATE", + (minute_ts,), + ) + row = await cur.fetchone() + if row is not None: + old_raw = ( + row["data"] + if isinstance(row, dict) + else row[0] + ) + merged = self._merge_metric_buckets( + json.loads(old_raw), data, + ) + await cur.execute( + "UPDATE system_metrics SET data = %s, " + "created_at = %s WHERE minute_ts = %s", + ( + json.dumps(merged, default=str), + now, + minute_ts, + ), + ) + else: + await cur.execute( + "INSERT INTO system_metrics " + "(minute_ts, data, created_at) " + "VALUES (%s, %s, %s)", + ( + minute_ts, + json.dumps(data, default=str), + now, + ), + ) + await conn.commit() + return + except Exception: + await conn.rollback() + raise + except (aiomysql.IntegrityError, aiomysql.OperationalError) as error: + code = error.args[0] if error.args else None + deadlock = code in (1205, 1213) + if attempt or (isinstance(error, aiomysql.OperationalError) and not deadlock): + raise + # A competing writer inserted the row, or InnoDB chose us + # as the deadlock victim. Retry so this batch merges. + continue except Exception as e: logger.warning(f"store_metrics_minute failed: {e}") diff --git a/tests/test_mysql_metrics_atomic.py b/tests/test_mysql_metrics_atomic.py new file mode 100644 index 0000000..adbe593 --- /dev/null +++ b/tests/test_mysql_metrics_atomic.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- +"""MySQL minute metrics must merge under one row lock.""" + +import asyncio +import json + +import pytest + +pytest.importorskip("aiomysql") + +from hy_memory.config import MemoryConfig # noqa: E402 +from hy_memory.data.cache_mysql import MysqlCache # noqa: E402 + + +class FakeCursor: + def __init__(self, row): + self.row = row + self.executed = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + async def execute(self, sql, args=None): + self.executed.append((sql, args)) + + async def fetchone(self): + return self.row + + +class FakeConnection: + def __init__(self, row): + self.cursor_obj = FakeCursor(row) + self.begun = False + self.committed = False + self.rolled_back = False + + async def begin(self): + self.begun = True + + async def commit(self): + self.committed = True + + async def rollback(self): + self.rolled_back = True + + def cursor(self, *args, **kwargs): + return self.cursor_obj + + +class _Acquire: + def __init__(self, connection): + self.connection = connection + + async def __aenter__(self): + return self.connection + + async def __aexit__(self, *args): + return None + + +class FakePool: + def __init__(self, connection): + self.connection = connection + + def acquire(self): + return _Acquire(self.connection) + + +def _cache_with(row): + cache = MysqlCache(MemoryConfig()) + connection = FakeConnection(row) + cache._pool = FakePool(connection) + return cache, connection + + +def test_metrics_insert_takes_the_row_lock_and_commits(): + cache, connection = _cache_with(None) + + asyncio.run( + cache.store_metrics_minute("2026-01-01T00:00", {"sys1_started": 2}) + ) + + statements = [sql for sql, _ in connection.cursor_obj.executed] + assert any("FOR UPDATE" in sql for sql in statements) + assert any(sql.startswith("INSERT") for sql in statements) + assert connection.begun and connection.committed + assert not connection.rolled_back + + +def test_metrics_update_merges_existing_buckets(): + old = {"sys1_started": 1, "sys1_timing_sums": {"a": 1}} + cache, connection = _cache_with({"data": json.dumps(old)}) + + asyncio.run( + cache.store_metrics_minute( + "2026-01-01T00:00", + {"sys1_started": 2, "sys1_timing_sums": {"a": 3}}, + ) + ) + + update = next( + args for sql, args in connection.cursor_obj.executed + if sql.startswith("UPDATE") + ) + merged = json.loads(update[0]) + assert merged["sys1_started"] == 3 + assert merged["sys1_timing_sums"]["a"] == 4 + assert connection.committed