modify: 分布式锁问题

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"