From c33ac236f6fc74a2000eb396f98d355e2e9c94e0 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 4 Sep 2026 21:22:34 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E5=88=86=E5=B8=83=E5=BC=8F=E9=94=81?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 13 +---- src/chongming_cache/backends/sqlite_.py | 9 +++ tests/test_cache.py | 75 +++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 10 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 5910aee..ce4e88e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,21 +4,14 @@ requires = ["hatchling"] [project] name = "chongming-cache" -version = "0.2.0" +version = "0.2.2" description = "统一缓存封装：兼容 sqlite(本地) / redis(共享) 后端" readme = "README.md" requires-python = ">=3.12" -dependencies = [ - "redis>=5.0.0", - "orjson>=3.9.0", -] +dependencies = ["redis>=5.0.0", "orjson>=3.9.0"] [dependency-groups] -dev = [ - "pytest>=8.0.0", - "fakeredis>=2.28.0", - "twine>=7.0.0", -] +dev = ["pytest>=8.0.0", "fakeredis>=2.28.0", "twine>=7.0.0"] [tool.hatch.build.targets.wheel] packages = ["src/chongming_cache"] diff --git a/src/chongming_cache/backends/sqlite_.py b/src/chongming_cache/backends/sqlite_.py index ce9cab4..ab4748f 100644 --- a/src/chongming_cache/backends/sqlite_.py +++ b/src/chongming_cache/backends/sqlite_.py @@ -140,6 +140,15 @@ class SqliteBackend(CacheBackend): def set_nx(self, key, value, expire: float | None = None) -> bool: with self._lock: k = str(key) + # 过期行先清除，再尝试插入（对齐 redis 后端 SET NX 语义：过期 key 自动 + # 让位）。否则持锁方崩溃后残留的过期锁行会让 INSERT OR IGNORE 永远失败， + # 分布式锁（inline 持有者选举 / MQTT Bridge 等）再也无法重新获取。 + # DELETE 只删已过期行，不会误删其他进程刚写入的新鲜行；并发下仍保证 + # 恰好一个 set_nx 成功（写锁串行化 + busy_timeout）。 + self._conn.execute( + "DELETE FROM cache WHERE k = ? AND expire IS NOT NULL AND expire <= ?", + (k, time.time()), + ) v = encode_value(value).decode("utf-8") cur = self._conn.execute( "INSERT OR IGNORE INTO cache(k, v, expire) VALUES(?, ?, ?)", diff --git a/tests/test_cache.py b/tests/test_cache.py index 425e0af..05b2557 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -178,6 +178,72 @@ def test_sqlite_json_interop_and_multi_instance(tmp_path): b.close() +# ── set_nx / 分布式锁租约语义（sqlite）──────────────────────────── +# 回归：持锁进程崩溃后残留的「已过期」锁行必须让位，否则后续 set_nx 永远失败， +# inline 持有者选举 / MQTT Bridge 等分布式锁死锁（生产单网关 sqlite 后端场景）。 + + +def test_set_nx_treats_expired_row_as_vacant(tmp_path): + """已过期条目（如崩溃残留的锁行）不得阻塞 set_nx（对齐 redis SET NX 语义）。""" + db = str(tmp_path / "nx.db") + cache = Cache.from_config({"backend": "sqlite", "path": db}) + try: + raw = sqlite3.connect(db) + raw.execute( + "INSERT INTO cache(k, v, expire) VALUES('lk', ?, ?)", + ('"old-token"', time.time() - 5), + ) + raw.commit() + raw.close() + assert cache.backend.set_nx("lk", "new-token", expire=30) is True + assert cache.get("lk") == "new-token" + finally: + cache.close() + + +def test_set_nx_live_row_blocks(tmp_path): + """未过期的存活条目仍保持互斥：第二次 set_nx 必须失败。""" + db = str(tmp_path / "nx2.db") + cache = Cache.from_config({"backend": "sqlite", "path": db}) + try: + assert cache.backend.set_nx("lk", "a", expire=60) is True + assert cache.backend.set_nx("lk", "b", expire=60) is False + assert cache.get("lk") == "a" + finally: + cache.close() + + +def test_distributed_lock_reacquire_after_crash_expiry(tmp_path): + """锁级回归：持有者崩溃（不释放、租约过期）后，新实例必须能重新获取锁。""" + from chongming_cache.lock import DistributedLock + + db = str(tmp_path / "lock.db") + cache = Cache.from_config({"backend": "sqlite", "path": db}) + try: + # 正常释放后立即可重新获取 + l1 = DistributedLock(cache, "inline_owner:test") + assert l1.acquire() is True + l1.release() + l2 = DistributedLock(cache, "inline_owner:test") + assert l2.acquire() is True + l2.release() + + # 崩溃场景：获取后不 release，直接把锁行置为已过期（模拟租约到期未续约） + crashed = DistributedLock(cache, "inline_owner:test") + assert crashed.acquire() is True + raw = sqlite3.connect(db) + raw.execute( + "UPDATE cache SET expire = ? WHERE k = 'inline_owner:test'", + (time.time() - 1,), + ) + raw.commit() + raw.close() + # 新进程（新 token）应重新获得锁；修复前 INSERT OR IGNORE 撞残留行恒失败 + assert DistributedLock(cache, "inline_owner:test").acquire() is True + finally: + cache.close() + + # ── redis 后端 ────────────────────────────────────────────────── @@ -229,3 +295,12 @@ def test_redis_binary_value_roundtrip(redis_cache): value = {"nested": {"list": [1, 2, 3]}, "bytes": b"\x00\x01"} redis_cache["k"] = value assert redis_cache["k"] == value + + +def test_set_nx_after_expiry_redis(redis_cache): + """redis 对照：过期 key 自动让位，set_nx 可再次成功；存活 key 保持互斥。""" + assert redis_cache.backend.set_nx("lk", "a", expire=1) is True + assert redis_cache.backend.set_nx("lk", "b", expire=10) is False + time.sleep(1.2) + assert redis_cache.backend.set_nx("lk", "c", expire=10) is True + assert redis_cache.get("lk") == "c"