基于 Python 的高性能内存键值存储引擎设计与实现

"""
高性能内存键值存储引擎
========================

模块覆盖：
1. 跳表（Skip List）索引结构
2. WAL 预写日志
3. LRU 缓存淘汰
4. 快照持久化与恢复
5. 简单事务（BEGIN/COMMIT/ROLLBACK）
6. 读写锁并发控制
7. TTL 过期清理
8. 监控统计
"""

import os
import sys
import json
import time
import random
import struct
import hashlib
import logging
import threading
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple, Callable
from pathlib import Path
from dataclasses import dataclass, field
from collections import OrderedDict

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("kvstore")


# ==================== 常量 ====================

DEFAULT_DATA_DIR = "./kvdata"
DEFAULT_WAL_FILE = "wal.log"
DEFAULT_SNAPSHOT_FILE = "snapshot.json"
DEFAULT_MAX_LEVEL = 16
DEFAULT_P_FACTOR = 0.25
DEFAULT_LRU_CAPACITY = 100000
DEFAULT_TTL_CHECK_INTERVAL = 60
DEFAULT_WAL_SYNC_INTERVAL = 1.0

# WAL 记录类型
WAL_OP_PUT = 1
WAL_OP_DELETE = 2
WAL_OP_TXN_BEGIN = 3
WAL_OP_TXN_COMMIT = 4
WAL_OP_TXN_ROLLBACK = 5


# ==================== 数据模型 ====================

@dataclass
class KVEntry: 2001.m.kuaisou.com 
    """键值条目"""
    key: str
    value: Any
    timestamp: float = field(default_factory=time.time)
    ttl: Optional[float] = None  # 秒，None 表示永不过期
    txn_id: Optional[str] = None
    deleted: bool = False

    @property
    def is_expired(self) -> bool: 2002.m.kuaisou.com 
        if self.ttl is None: 2005.m.kuaisou.com 
            return False
        return time.time() > self.timestamp + self.ttl

    def to_dict(self) -> Dict[str, Any]: 2004.m.kuaisou.com 
        return {
            "key": self.key,
            "value": self.value,
            "timestamp": self.timestamp,
            "ttl": self.ttl,
            "deleted": self.deleted,
        }

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "KVEntry":
        return cls(**data)


@dataclass
class WALRecord: 2006.m.kuaisou.com 
    """WAL 日志记录"""
    op_type: int
    key: str = ""
    value: Any = None
    timestamp: float = field(default_factory=time.time)
    txn_id: Optional[str] = None
    ttl: Optional[float] = None
    checksum: str = ""

    def serialize(self) -> bytes: 2008.m.kuaisou.com 
        payload = json.dumps({
            "op": self.op_type,
            "key": self.key,
            "value": self.value,
            "ts": self.timestamp,
            "txn": self.txn_id,
            "ttl": self.ttl,
        }, default=str).encode("utf-8")
        checksum = hashlib.md5(payload).hexdigest()[:8]
        length = len(payload)
        header = struct.pack("!I8s", length, checksum.encode("ascii"))
        return header + payload

    @classmethod
    def deserialize(cls, data: bytes) -> "WALRecord": 2009.m.kuaisou.com 
        length, checksum_bytes = struct.unpack("!I8s", data[:12])
        payload = data[12:12 + length]
        expected_checksum = checksum_bytes.decode("ascii").rstrip("\x00")
        actual_checksum = hashlib.md5(payload).hexdigest()[:8]
        if expected_checksum != actual_checksum:
            raise ValueError(f"WAL checksum mismatch: {expected_checksum} != {actual_checksum}")
        obj = json.loads(payload.decode("utf-8"))
        return cls(
            op_type=obj["op"],
            key=obj.get("key", ""),
            value=obj.get("value"),
            timestamp=obj.get("ts", 0),
            txn_id=obj.get("txn"),
            ttl=obj.get("ttl"),
            checksum=expected_checksum,
        )


# ==================== 跳表节点 ====================

class SkipListNode: 2010.m.kuaisou.com 
    """跳表节点"""
    __slots__ = ("key", "entry", "forward")

    def __init__(self, key: str = "", entry: Optional[KVEntry] = None, level: int = 0):
        self.key = key
        self.entry = entry
        self.forward: List[Optional["SkipListNode"]] = [None] * (level + 1)


class SkipList: 2011.m.kuaisou.com 
    """跳表索引"""

    def __init__(self, max_level: int = DEFAULT_MAX_LEVEL, p: float = DEFAULT_P_FACTOR):
        self.max_level = max_level
        self.p = p
        self.header = SkipListNode(level=max_level)
        self.level = 0
        self.size = 0
        self._lock = threading.RLock()

    def _random_level(self) -> int: 2012.m.kuaisou.com 
        lvl = 0
        while random.random() < self.p and lvl < self.max_level: 2013.m.kuaisou.com
            lvl += 1
        return lvl

    def insert(self, key: str, entry: KVEntry) -> bool:
        with self._lock: 2014.m.kuaisou.com 
            update = [None] * (self.max_level + 1)
            current = self.header

            for i in range(self.level, -1, -1): 2016.m.kuaisou.com 
                while (current.forward[i] is not None and
                       current.forward[i].key < key): 2017.m.kuaisou.com 
                    current = current.forward[i]
                update[i] = current

            next_node = current.forward[0]
            if next_node is not None and next_node.key == key: 2018.m.kuaisou.com 
                next_node.entry = entry
                return False

            new_level = self._random_level(2020.m.kuaisou.com)
            if new_level > self.level: 2021.m.kuaisou.com 
                for i in range(self.level + 1, new_level + 1): 2022.m.kuaisou.com 
                    update[i] = self.header
                self.level = new_level

            new_node = SkipListNode(key=key, entry=entry, level=new_level)
            for i in range(new_level + 1):
                new_node.forward[i] = update[i].forward[i]
                update[i].forward[i] = new_node

            self.size += 1
            return True

    def search(self, key: str) -> Optional[KVEntry]: 2024.m.kuaisou.com 
        with self._lock: 2023.m.kuaisou.com 
            current = self.header
            for i in range(self.level, -1, -1): 2028.m.kuaisou.com 
                while (current.forward[i] is not None and
                       current.forward[i].key < key): 2027.m.kuaisou.com 
                    current = current.forward[i]
            current = current.forward[0]
            if current is not None and current.key == key: 2030.m.kuaisou.com 
                return current.entry
            return None

    def delete(self, key: str) -> Optional[KVEntry]: 2029.m.kuaisou.com 
        with self._lock:
            update = [None] * (self.max_level + 1)
            current = self.header

            for i in range(self.level, -1, -1): 2030.m.kuaisou.com 
                while (current.forward[i] is not None and
                       current.forward[i].key < key): 2031.m.kuaisou.com 
                    current = current.forward[i]
                update[i] = current

            target = current.forward[0]
            if target is None or target.key != key: 2032.m.kuaisou.com 
                return None

            entry = target.entry
            for i in range(self.level + 1): 2034.m.kuaisou.com 
                if update[i].forward[i] != target: 2035.m.kuaisou.com 
                    break
                update[i].forward[i] = target.forward[i]

            while self.level > 0 and self.header.forward[self.level] is None:
                self.level -= 1

            self.size -= 1
            return entry

    def range_scan(self, start_key: str, end_key: str) -> List[KVEntry]:
        with self._lock: 2036.m.kuaisou.com 
            results = []
            current = self.header
            for i in range(self.level, -1, -1): 2037.m.kuaisou.com 
                while (current.forward[i] is not None and
                       current.forward[i].key < start_key): 2039.m.kuaisou.com 
                    current = current.forward[i]
            current = current.forward[0]
            while current is not None and current.key <= end_key: 2040.m.kuaisou.com 
                if current.entry and not current.entry.deleted and not current.entry.is_expired:
                    results.append(current.entry)
                current = current.forward[0]
            return results

    def all_entries(self) -> List[KVEntry]: 2041.m.kuaisou.com 
        with self._lock:
            entries = []
            current = self.header.forward[0]
            while current is not None: 2042.m.kuaisou.com 
                if current.entry and not current.entry.deleted:
                    entries.append(current.entry)
                current = current.forward[0]
            return entries


# ==================== LRU 缓存 ====================

class LRUCache: 2043.m.kuaisou.com 
    """LRU 淘汰缓存"""

    def __init__(self, capacity: int = DEFAULT_LRU_CAPACITY): 3010.m.kuaisou.com 
        self.capacity = capacity
        self._cache: OrderedDict[str, KVEntry] = OrderedDict()
        self._lock = threading.Lock(3015.m.kuaisou.com)
        self.hits = 0
        self.misses = 0

    @property
    def hit_rate(self) -> float: 4005.m.kuaisou.com 
        total = self.hits + self.misses
        return self.hits / total if total > 0 else 0.0

    def get(self, key: str) -> Optional[KVEntry]: 4006.m.kuaisou.com 
        with self._lock: 4012.m.kuaisou.com 
            if key in self._cache: 5002.m.kuaisou.com 
                self._cache.move_to_end(key)
                entry = self._cache[key]
                if entry.is_expired: 5007.m.kuaisou.com 
                    del self._cache[key]
                    self.misses += 1
                    return None
                self.hits += 1
                return entry
            self.misses += 1
            return None

    def put(self, key: str, entry: KVEntry): 5011.m.kuaisou.com 
        with self._lock: 5030.m.kuaisou.com 
            if key in self._cache: 5034.m.kuaisou.com 
                self._cache.move_to_end(key)
                self._cache[key] = entry
            else: 5052.m.kuaisou.com 
                if len(self._cache) >= self.capacity: 5038.m.kuaisou.com 
                    self._cache.popitem(last=False)
                self._cache[key] = entry

    def remove(self, key: str) -> Optional[KVEntry]:
        with self._lock: 5055.m.kuaisou.com 
            return self._cache.pop(key, None)

    def clear(self): 5057.m.kuaisou.com 
        with self._lock: 6004.m.kuaisou.com 
            self._cache.clear()
            self.hits = 0
            self.misses = 0


# ==================== WAL 管理器 ====================

class WALManager: 6007.m.kuaisou.com 
    """预写日志管理器"""

    def __init__(self, wal_path: Path): 6008.m.kuaisou.com 
        self.wal_path = wal_path
        self._lock = threading.Lock()
        self._buffer: List[WALRecord] = []
        self._sync_interval = DEFAULT_WAL_SYNC_INTERVAL
        self._last_sync = time.time()

    def append(self, record: WALRecord):
        with self._lock: 14009.m.kuaisou.com 
            self._buffer.append(record)
            now = time.time()
            if now - self._last_sync >= self._sync_interval:
                self._flush_locked()

    def flush(self): 14013.m.kuaisou.com 
        with self._lock: 14068.m.kuaisou.com 
            self._flush_locked()

    def _flush_locked(self): 14070.m.kuaisou.com 
        if not self._buffer: 14071.m.kuaisou.com 
            return
        try:
            with open(self.wal_path, "ab") as f:
                for record in self._buffer: 14080.m.kuaisou.com 
                    f.write(record.serialize())
                f.flush(14083.m.kuaisou.com)
                os.fsync(f.fileno())
            self._buffer.clear()
            self._last_sync = time.time(14082.m.kuaisou.com)
        except Exception as e:
            logger.error(f"WAL flush failed: {e}")

    def replay(self) -> List[WALRecord]: 14091.m.kuaisou.com 
        records = []
        if not self.wal_path.exists(): 14098.m.kuaisou.com 
            return records
        try: 14101.m.kuaisou.com 
            with open(self.wal_path, "rb") as f: 14100.m.kuaisou.com 
                while True:
                    header = f.read(12)
                    if len(header) < 12: 14102.m.kuaisou.com 
                        break
                    length, _ = struct.unpack("!I8s", header)
                    payload = f.read(length)
                    if len(payload) < length: 14125.m.kuaisou.com 
                        break
                    record = WALRecord.deserialize(header + payload)
                    records.append(record)
        except Exception as e: 14133.m.kuaisou.com 
            logger.error(f"WAL replay failed: {e}")
        return records

    def truncate(self): 14383.m.kuaisou.com 
        with self._lock: 14145.m.kuaisou.com 
            self._buffer.clear()
            if self.wal_path.exists(): 14384.m.kuaisou.com 
                self.wal_path.unlink()


# ==================== 事务管理器 ====================

class TransactionState(Enum): 15002.m.kuaisou.com 
    ACTIVE = "active"
    COMMITTED = "committed"
    ABORTED = "aborted"


@dataclass
class Transaction: 15015.m.kuaisou.com 
    txn_id: str
    state: TransactionState = TransactionState.ACTIVE
    write_set: Dict[str, KVEntry] = field(default_factory=dict)
    delete_set: set = field(default_factory=set)
    read_set: set = field(default_factory=set)
    start_time: float = field(default_factory=time.time)


class TransactionManager:
    """事务管理器"""

    def __init__(self): 15025.m.kuaisou.com 
        self._transactions: Dict[str, Transaction] = {}
        self._lock = threading.Lock()

    def begin(self, txn_id: str) -> Transaction:
        with self._lock: 15027.m.kuaisou.com 
            txn = Transaction(txn_id=txn_id)
            self._transactions[txn_id] = txn
            return txn

    def get(self, txn_id: str) -> Optional[Transaction]:
        with self._lock: 15029.m.kuaisou.com 
            return self._transactions.get(txn_id)

    def commit(self, txn_id: str) -> bool:
        with self._lock: 15030.m.kuaisou.com 
            txn = self._transactions.get(txn_id)
            if not txn or txn.state != TransactionState.ACTIVE:
                return False
            txn.state = TransactionState.COMMITTED
            return True

    def rollback(self, txn_id: str) -> bool:
        with self._lock: 15031.m.kuaisou.com 
            txn = self._transactions.get(txn_id)
            if not txn or txn.state != TransactionState.ACTIVE:
                return False
            txn.state = TransactionState.ABORTED
            return True

    def cleanup(self, txn_id: str):
        with self._lock: 15032.m.kuaisou.com 
            self._transactions.pop(txn_id, None)


# ==================== KV 存储引擎主体 ====================

class KVStore: 15033.m.kuaisou.com 
    """高性能内存键值存储引擎"""

    def __init__(
        self,
        data_dir: str = DEFAULT_DATA_DIR,
        lru_capacity: int = DEFAULT_LRU_CAPACITY,
        max_level: int = DEFAULT_MAX_LEVEL,
        wal_sync_interval: float = DEFAULT_WAL_SYNC_INTERVAL,
    ):
        self.data_dir = Path(data_dir)
        self.data_dir.mkdir(parents=True, exist_ok=True)

        self.index = SkipList(max_level=max_level)
        self.cache = LRUCache(capacity=lru_capacity)
        self.wal = WALManager(self.data_dir / DEFAULT_WAL_FILE)
        self.wal._sync_interval = wal_sync_interval
        self.txn_manager = TransactionManager(15034.m.kuaisou.com)

        self._rw_lock = threading.RLock(15035.m.kuaisou.com)
        self._ttl_thread: Optional[threading.Thread] = None
        self._running = False

        self.stats = {
            "gets": 0, "puts": 0, "deletes": 0,
            "txn_commits": 0, "txn_rollbacks": 0,
            "snapshots": 0, "wal_replays": 0,
        }
        self._stats_lock = threading.Lock(15036.m.kuaisou.com)

        self._recover(15038.m.kuaisou.com)

    def _inc_stat(self, key: str, delta: int = 1): 15037.m.kuaisou.com 
        with self._stats_lock:
            self.stats[key] = self.stats.get(key, 0) + delta

    # ---------- 核心 CRUD ----------

    def put(self, key: str, value: Any, ttl: Optional[float] = None, txn_id: Optional[str] = None):
        entry = KVEntry(key=key, value=value, ttl=ttl, txn_id=txn_id)

        if txn_id:
            txn = self.txn_manager.get(txn_id)
            if not txn or txn.state != TransactionState.ACTIVE: 15039.m.kuaisou.com 
                raise ValueError(f"Invalid transaction: {txn_id}")
            txn.write_set[key] = entry
            txn.delete_set.discard(key)
            self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, txn_id=txn_id, ttl=ttl))
            return

        with self._rw_lock: 16001.m.kuaisou.com 
            self.index.insert(key, entry)
            self.cache.put(key, entry)
            self.wal.append(WALRecord(op_type=WAL_OP_PUT, key=key, value=value, ttl=ttl))
            self._inc_stat("puts")

    def get(self, key: str, txn_id: Optional[str] = None) -> Optional[Any]:
        self._inc_stat("gets")

        if txn_id: 16002.m.kuaisou.com 
            txn = self.txn_manager.get(txn_id)
            if txn and txn.state == TransactionState.ACTIVE: 16004.m.kuaisou.com 
                if key in txn.delete_set: 16003.m.kuaisou.com 
                    return None
                if key in txn.write_set: 16005.m.kuaisou.com 
                    return txn.write_set[key].value
                txn.read_set.add(key)

        cached = self.cache.get(key)
        if cached is not None: 16006.m.kuaisou.com 
            return cached.value

        with self._rw_lock: 16007.m.kuaisou.com 
            entry = self.index.search(key)
            if entry is None or entry.deleted or entry.is_expired:
                return None
            self.cache.put(key, entry)
            return entry.value

    def delete(self, key: str, txn_id: Optional[str] = None) -> bool:
        if txn_id: 16008.m.kuaisou.com 
            txn = self.txn_manager.get(txn_id)
            if not txn or txn.state != TransactionState.ACTIVE:
                raise ValueError(f"Invalid transaction: {txn_id}")
            txn.delete_set.add(key)
            txn.write_set.pop(key, None)
            self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key, txn_id=txn_id))
            return True

        with self._rw_lock: 16009.m.kuaisou.com 
            entry = self.index.delete(key)
            self.cache.remove(key)
            self.wal.append(WALRecord(op_type=WAL_OP_DELETE, key=key))
            self._inc_stat("deletes")
            return entry is not None

    def exists(self, key: str) -> bool: 16010.m.kuaisou.com 
        return self.get(key) is not None

    def range_query(self, start_key: str, end_key: str) -> List[Tuple[str, Any]]:
        with self._rw_lock: 16011.m.kuaisou.com 
            entries = self.index.range_scan(start_key, end_key)
            return [(e.key, e.value) for e in entries]

    # ---------- 事务 ----------

    def begin_transaction(self) -> str: 16012.m.kuaisou.com 
        txn_id = hashlib.md5(f"{time.time()}{random.random()}".encode()).hexdigest()[:12]
        self.txn_manager.begin(txn_id)
        self.wal.append(WALRecord(op_type=WAL_OP_TXN_BEGIN, txn_id=txn_id))
        return txn_id

    def commit_transaction(self, txn_id: str) -> bool: 16013.m.kuaisou.com 
        txn = self.txn_manager.get(txn_id)
        if not txn or txn.state != TransactionState.ACTIVE: 16014.m.kuaisou.com 
            return False

        with self._rw_lock: 16016.m.kuaisou.com 
            for key, entry in txn.write_set.items(): 16015.m.kuaisou.com 
                self.index.insert(key, entry)
                self.cache.put(key, entry)
            for key in txn.delete_set:
                self.index.delete(key)
                self.cache.remove(key)

        self.txn_manager.commit(txn_id)
        self.wal.append(WALRecord(op_type=WAL_OP_TXN_COMMIT, txn_id=txn_id))
        self.txn_manager.cleanup(txn_id)
        self._inc_stat("txn_commits")
        return True

    def rollback_transaction(self, txn_id: str) -> bool: 16017.m.kuaisou.com 
        result = self.txn_manager.rollback(txn_id)
        if result: 16018.m.kuaisou.com 
            self.wal.append(WALRecord(op_type=WAL_OP_TXN_ROLLBACK, txn_id=txn_id))
            self.txn_manager.cleanup(txn_id)
            self._inc_stat("txn_rollbacks")
        return result

    # ---------- 持久化 ----------

    def snapshot(self) -> str:
        with self._rw_lock: 16019.m.kuaisou.com 
            entries = self.index.all_entries()
            data = [e.to_dict() for e in entries if not e.is_expired]

        snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE
        tmp_path = snapshot_path.with_suffix(".tmp")
        try: 16020.m.kuaisou.com 
            tmp_path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8")
            tmp_path.replace(snapshot_path)
            self.wal.truncate()
            self._inc_stat("snapshots")
            logger.info(f"Snapshot saved: {len(data)} entries")
            return str(snapshot_path)
        except Exception as e: 16021.m.kuaisou.com 
            logger.error(f"Snapshot failed: {e}")
            raise

    def _recover(self): 16022.m.kuaisou.com 
        snapshot_path = self.data_dir / DEFAULT_SNAPSHOT_FILE
        if snapshot_path.exists():
            try: 16023.m.kuaisou.com 
                data = json.loads(snapshot_path.read_text(encoding="utf-8"))
                for item in data: 16024.m.kuaisou.com 
                    entry = KVEntry.from_dict(item)
                    if not entry.is_expired: 16025.m.kuaisou.com 
                        self.index.insert(entry.key, entry)
                        self.cache.put(entry.key, entry)
                logger.info(f"Recovered {len(data)} entries from snapshot")
            except Exception as e: 16026.m.kuaisou.com 
                logger.error(f"Snapshot recovery failed: {e}")

        wal_records = self.wal.replay(16027.m.kuaisou.com)
        if wal_records:
            txns: Dict[str, Dict] = {}
            for record in wal_records: 16028.m.kuaisou.com 
                if record.op_type == WAL_OP_PUT: 16029.m.kuaisou.com 
                    if record.txn_id: 16030.m.kuaisou.com 
                        txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()})
                        txns[record.txn_id]["writes"][record.key] = KVEntry(
                            key=record.key, value=record.value, ttl=record.ttl, txn_id=record.txn_id
                        )
                    else: 16032.m.kuaisou.com 
                        entry = KVEntry(key=record.key, value=record.value, ttl=record.ttl)
                        self.index.insert(record.key, entry)
                        self.cache.put(record.key, entry)
                elif record.op_type == WAL_OP_DELETE: 16031.m.kuaisou.com 
                    if record.txn_id: 16033.m.kuaisou.com 
                        txns.setdefault(record.txn_id, {"writes": {}, "deletes": set()})
                        txns[record.txn_id]["deletes"].add(record.key)
                    else: 16034.m.kuaisou.com 
                        self.index.delete(record.key)
                        self.cache.remove(record.key)
                elif record.op_type == WAL_OP_TXN_COMMIT:
                    txn_data = txns.pop(record.txn_id, None)
                    if txn_data: 16035.m.kuaisou.com 
                        for k, v in txn_data["writes"].items():
                            self.index.insert(k, v)
                            self.cache.put(k, v)
                        for k in txn_data["deletes"]:
                            self.index.delete(k)
                            self.cache.remove(k)
                elif record.op_type == WAL_OP_TXN_ROLLBACK:
                    txns.pop(record.txn_id, None)

            self._inc_stat("wal_replays", len(wal_records))
            logger.info(f"Replayed {len(wal_records)} WAL records")

    # ---------- TTL 清理 ----------

    def start_ttl_checker(self, interval: float = DEFAULT_TTL_CHECK_INTERVAL):
        self._running = True
        self._ttl_thread = threading.Thread(target=self._ttl_loop, args=(interval,), daemon=True)
        self._ttl_thread.start(16036.m.kuaisou.com)
        logger.info(f"TTL checker started (interval={interval}s)")

    def stop_ttl_checker(self): 16037.m.kuaisou.com 
        self._running = False
        if self._ttl_thread:
            self._ttl_thread.join(timeout=5.0)

    def _ttl_loop(self, interval: float): 16038.m.kuaisou.com 
        while self._running: 16039.m.kuaisou.com 
            time.sleep(interval)
            self._cleanup_expired()

    def _cleanup_expired(self):
        with self._rw_lock: 16040.m.kuaisou.com 
            entries = self.index.all_entries()
            expired_keys = [e.key for e in entries if e.is_expired]
            for key in expired_keys: 16041.m.kuaisou.com 
                self.index.delete(key)
                self.cache.remove(key)
            if expired_keys:
                logger.debug(f"Cleaned up {len(expired_keys)} expired keys")

    # ---------- 监控 ----------

    def get_stats(self) -> Dict[str, Any]:
        with self._stats_lock: 16042.m.kuaisou.com 
            stats = dict(self.stats)
        stats["index_size"] = self.index.size
        stats["cache_size"] = len(self.cache._cache)
        stats["cache_hit_rate"] = round(self.cache.hit_rate, 4)
        return stats

    def close(self): 16043.m.kuaisou.com 
        self.stop_ttl_checker()
        self.wal.flush()
        logger.info("KVStore closed")