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 <shijiahao_0802@youzan.com>
Co-authored-by: Claude Code <noreply@anthropic.com>

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}")
 

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