migration 'chat_prekey' in lmdb binary ( node relay ), and fixing some perfomance issue in chat e2ee.

Author: Tsarstd

src/tsarchain/network/node_logic/chat_state.py

@@ -12,8 +12,8 @@
 def init_chat_state(self) -> None:
     self.chat_lock = threading.RLock()
     self.chat_presence_pub = {}
+    self.chat_presence_ts = {}
     self.chat_spend_pub = {}
-    self.chat_prekeys = {}
     self.chat_mailbox = {}
     self.chat_global_count = 0
     self.chat_presence_seen = set()

src/tsarchain/network/rpc/docs/USER_RPC.MD

@@ -288,16 +288,17 @@ Provides decentralized, end-to-end encrypted messaging via TsarChain's Signal-co
 The node operates purely as a **Blind Relay & Ephemeral Mailbox Router** with zero ability to decrypt ciphertexts and zero permanent disk persistence for transient chat messages:
 
 #### 1. Prekey Bundle Management (`chat_prekeys`)
-* **Data Structure**: In-memory dictionary `self.chat_prekeys[addr]`.
+* **Data Structure**: Dedicated LMDB binary database `LMDB_CHAT_PREKEYS` (`data/node/chat_prekeys`).
 * **Bundle Components**:
   - `ik` *(Identity Key)*: User's long-term public X25519 identity key.
   - `spk` *(Signed Prekey)*: Semi-static ephemeral public key authenticated via ECDSA Secp256k1.
   - `sig`: Identity digital signature over the canonical payload `TSAR-SPK|<spk_bytes>|<spend_pub_bytes>`.
   - `opk_list` *(One-Time Prekeys Pool)*: Pool of single-use ephemeral keys for maximum Forward Secrecy.
+* **Compact Binary Encoding**: Packed into a zero-overhead raw binary structure (`<QB` header, 32B IK, 32B SPK, signature bytes, and contiguous 32B OPKs) for >60% storage reduction.
 * **Atomic Key Consumption (FIFO Pop)**:
-  - When an initiator calls `CHAT_GET_PREKEY`, the node consumes the top OPK via `lst.pop(0)`.
+  - When an initiator calls `CHAT_GET_PREKEY`, the node consumes the top OPK via `lst.pop(0)` and atomically updates the record in LMDB.
   - A consumed OPK is **never issued to any other peer**, ensuring each new cryptographic session begins with unique ephemeral entropy.
-* **Capacity Constraints**: Capped at `CFG.CHAT_OPK_MAX_STORED` (default: 100 OPKs per address) to prevent memory exhaustion.
+* **Capacity Management**: Backed by `CFG.LMDB_PREKEYS_SIZE_MAX = 250MB` with rate limiting and PoW verification on publish endpoints.
 
 #### 2. Offline Message Mailbox Queue Management (`chat_mailbox`)
 * **Data Structure**: In-memory FIFO queue per recipient: `self.chat_mailbox[addr] = collections.deque()`.

src/tsarchain/network/rpc/user_rpc/category/chat.py

@@ -106,27 +106,30 @@ def chat_register(self, message, pow_obj, base_identity, addr, *,
             return spk_err
 
         now = time.time()
+        now_int = int(now)
         pid = secrets.token_hex(16)
         with self.chat_lock:
             self.chat_spend_pub[addr_s] = spend_pk
             self.chat_presence_pub[addr_s] = chat_pub
+            if hasattr(self, "chat_presence_ts"):
+                self.chat_presence_ts[addr_s] = now_int
             if hasattr(self, "record_presence_seen"):
                 self.record_presence_seen(pid)
             else:
                 self.chat_presence_seen.add(pid)
-            b = self.chat_prekeys.get(addr_s) or {}
+            b = self.get_prekey_bundle(addr_s)
             if "ik" not in b: 
                 b["ik"] = chat_pub
-                b["ts"] = int(now)
-            b["ts"] = int(now)
+                b["ts"] = now_int
+            b["ts"] = now_int
             if spk_valid:
                 b["spk"] = spk_reg
                 b["sig"] = sig_reg
             if opk_reg and len(opk_reg) == 64:
                 b.setdefault("opk_list", []).append(opk_reg)
-            self.chat_prekeys[addr_s] = b
+            self.put_prekey_bundle(addr_s, b)
 
-        pres = {"pid": pid, "address": addr_s, "pubkey": chat_pub, "spend_pub": spend_pk, "presence_sig": presence_sig, "ts": int(now), "hops": 0}
+        pres = {"pid": pid, "address": addr_s, "pubkey": chat_pub, "spend_pub": spend_pk, "presence_sig": presence_sig, "ts": now_int, "hops": 0}
         self.relay_presence_async(pres, exclude=addr)
         return {"type": "CHAT_REGISTERED", "address": addr_s, "pubkey": chat_pub}
     except Exception as exc:
@@ -175,10 +178,15 @@ def chat_lookup_pub(self, message, pow_obj, base_identity, *,
         return pow_resp
     pubhex = self.chat_presence_pub.get(addr_s)
     last_seen = None
-    b = self.chat_prekeys.get(addr_s) or {}
-    ts_field = b.get("ts")
-    if isinstance(ts_field, (int, float)):
-        last_seen = int(ts_field)
+    if hasattr(self, "chat_presence_ts"):
+        last_seen = self.chat_presence_ts.get(addr_s)
+    if last_seen is None:
+        b = self.get_prekey_bundle(addr_s)
+        ts_field = b.get("ts")
+        if isinstance(ts_field, (int, float)):
+            last_seen = int(ts_field)
+            if hasattr(self, "chat_presence_ts"):
+                self.chat_presence_ts[addr_s] = last_seen
 
     return {"type": "CHAT_PUBKEY", "address": addr_s, "pubkey": pubhex, "found": bool(pubhex), "last_seen": last_seen}
 
@@ -237,18 +245,21 @@ def chat_presence(self, message, pow_obj, base_identity, addr, *,
         return pow_resp
 
     pid = message.get("pid") or secrets.token_hex(16)
+    now_int = int(time.time())
     with self.chat_lock:
         self.chat_presence_pub[addr_s] = pubhex
         self.chat_spend_pub[addr_s] = spend_pk
+        if hasattr(self, "chat_presence_ts"):
+            self.chat_presence_ts[addr_s] = now_int
         if hasattr(self, "record_presence_seen"):
             self.record_presence_seen(pid)
         else:
             self.chat_presence_seen.add(pid)
-        b = self.chat_prekeys.get(addr_s) or {}
+        b = self.get_prekey_bundle(addr_s)
         if "ik" not in b:
             b["ik"] = pubhex
-        b["ts"] = int(time.time())
-        self.chat_prekeys[addr_s] = b
+        b["ts"] = now_int
+        self.put_prekey_bundle(addr_s, b)
 
     message["hops"] = hops + 1
     self.relay_presence_async(message, exclude=addr)
@@ -294,16 +305,15 @@ def chat_publish_prekeys(self, message, pow_obj, base_identity, *,
     if not sig_ok.get("spk"):
         log.warning("[chat_publish_prekeys] Bad SPK signature for %s", addr_s)
         return {"error":"bad_spk_sig"}
+    now_int = int(time.time())
     with self.chat_lock:
-        rec = self.chat_prekeys.get(addr_s) or {}
-        rec.update({"ik": ik, "spk": spk, "sig": sig, "ts": int(time.time())})
-        if isinstance(opk, str) and len(opk)==64:
-            lst = rec.setdefault("opk_list", [])
-            lst.append(opk)
-            if len(lst) > CFG.CHAT_OPK_MAX_STORED:
-                # keep it from getting bloated, only keep the latest OPK
-                rec["opk_list"] = lst[-CFG.CHAT_OPK_MAX_STORED:]
-        self.chat_prekeys[addr_s] = rec
+        if hasattr(self, "chat_presence_ts"):
+            self.chat_presence_ts[addr_s] = now_int
+        rec = self.get_prekey_bundle(addr_s)
+        rec.update({"ik": ik, "spk": spk, "sig": sig, "ts": now_int})
+        if isinstance(opk, str) and len(opk) == 64:
+            rec.setdefault("opk_list", []).append(opk)
+        self.put_prekey_bundle(addr_s, rec)
 
     return {"type":"CHAT_PUBLISH_PREKEYS"}
 
@@ -313,12 +323,16 @@ def chat_get_prekey(self, message, *,
                     client_ip, is_miner_sender, **kwargs):
     addr_s = (message.get("address") or "").strip().lower()
     with self.chat_lock:
-        b = self.chat_prekeys.get(addr_s) or {}
+        b = self.get_prekey_bundle(addr_s)
         if not b or ("ik" not in b or "spk" not in b or "sig" not in b):
             return {"error":"no_bundle"}
         lst = b.get("opk_list") or []
         opk = lst.pop(0) if lst else None
-        self.chat_prekeys[addr_s] = b
+        if lst:
+            b["opk_list"] = lst
+        else:
+            b.pop("opk_list", None)
+        self.put_prekey_bundle(addr_s, b)
         sp = self.chat_spend_pub.get(addr_s)
 
     return {"type":"CHAT_PREKEY_BUNDLE","bundle":{"ik": b["ik"], "spk": b["spk"], "sig": b["sig"], "opk": opk, "spend_pub": sp}}

src/tsarchain/network/rpc_helper/chat.py

@@ -6,22 +6,150 @@
 import json
 import time
 import socket
+import struct
 import threading
 import collections
+from concurrent.futures import ThreadPoolExecutor
 
 # ---------------- Local Project ----------------
 from ...utils import config as CFG
 from .base import NetworkHandlerProxy
 from ..protocol import send_message, recv_message,build_envelope, SecureChannel
+from ...storage.kv import get as kv_get, put as kv_put, delete as kv_delete
 
 # ---------------- Logger ----------------
 from ...utils.tsar_logging import get_ctx_logger
 log = get_ctx_logger("tsarchain.network.rpc_helper.chat")
 
+_presence_executor: ThreadPoolExecutor | None = None
+_presence_executor_lock = threading.Lock()
+
+
+def get_presence_executor() -> ThreadPoolExecutor:
+    global _presence_executor
+    if _presence_executor is None:
+        with _presence_executor_lock:
+            if _presence_executor is None:
+                _presence_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="presence_relay")
+    return _presence_executor
+
+
+def encode_prekey_bundle(bundle: dict) -> bytes:
+    if not isinstance(bundle, dict):
+        return b""
+    ts = int(bundle.get("ts", 0) or 0)
+    ik = bundle.get("ik")
+    spk = bundle.get("spk")
+    sig = bundle.get("sig")
+    opk_list = bundle.get("opk_list") or []
+
+    flags = 0
+    body = bytearray()
+    if isinstance(ik, str) and len(ik) == 64:
+        try:
+            body.extend(bytes.fromhex(ik))
+            flags |= 0x01
+        except ValueError:
+            pass
+    if isinstance(spk, str) and len(spk) == 64:
+        try:
+            body.extend(bytes.fromhex(spk))
+            flags |= 0x02
+        except ValueError:
+            pass
+    if isinstance(sig, str) and sig:
+        try:
+            sig_bytes = bytes.fromhex(sig)
+            if len(sig_bytes) <= 65535:
+                body.extend(struct.pack("<H", len(sig_bytes)))
+                body.extend(sig_bytes)
+                flags |= 0x04
+        except ValueError:
+            pass
+
+    opk_bytes = bytearray()
+    if isinstance(opk_list, list):
+        for o in opk_list:
+            if isinstance(o, str) and len(o) == 64:
+                try:
+                    opk_bytes.extend(bytes.fromhex(o))
+                except ValueError:
+                    pass
+    opk_count = len(opk_bytes) // 32
+    opk_header = struct.pack("<I", opk_count)
+    header = struct.pack("<QB", ts, flags)
+    return bytes(header + body + opk_header + opk_bytes)
+
+
+def decode_prekey_bundle(raw: bytes) -> dict:
+    if not raw or len(raw) < 9:
+        return {}
+    try:
+        ts, flags = struct.unpack_from("<QB", raw, 0)
+        offset = 9
+        ik = None
+        spk = None
+        sig = None
+        if flags & 0x01:
+            if offset + 32 <= len(raw):
+                ik = raw[offset:offset+32].hex()
+                offset += 32
+        if flags & 0x02:
+            if offset + 32 <= len(raw):
+                spk = raw[offset:offset+32].hex()
+                offset += 32
+        if flags & 0x04:
+            if offset + 2 <= len(raw):
+                sig_len = struct.unpack_from("<H", raw, offset)[0]
+                offset += 2
+                if offset + sig_len <= len(raw):
+                    sig = raw[offset:offset+sig_len].hex()
+                    offset += sig_len
+        opk_list = []
+        if offset + 4 <= len(raw):
+            opk_count = struct.unpack_from("<I", raw, offset)[0]
+            offset += 4
+            for _ in range(opk_count):
+                if offset + 32 <= len(raw):
+                    opk_list.append(raw[offset:offset+32].hex())
+                    offset += 32
+        res: dict = {"ts": ts}
+        if ik:
+            res["ik"] = ik
+        if spk:
+            res["spk"] = spk
+        if sig:
+            res["sig"] = sig
+        if opk_list:
+            res["opk_list"] = opk_list
+        return res
+    except Exception:
+        return {}
+
 
 # ------------------------------ P2P Chat ------------------------------
 
 class ChatHandler(NetworkHandlerProxy):
+    def get_prekey_bundle(self, addr: str) -> dict:
+        if not addr:
+            return {}
+        with self.chat_lock:
+            raw = kv_get("chat_prekeys", addr.encode("utf-8"))
+            return decode_prekey_bundle(raw) if raw else {}
+
+    def put_prekey_bundle(self, addr: str, bundle: dict) -> None:
+        if not addr:
+            return
+        with self.chat_lock:
+            raw = encode_prekey_bundle(bundle)
+            kv_put("chat_prekeys", addr.encode("utf-8"), raw)
+
+    def delete_prekey_bundle(self, addr: str) -> None:
+        if not addr:
+            return
+        with self.chat_lock:
+            kv_delete("chat_prekeys", addr.encode("utf-8"))
+
     def send_to_peer(self, peer: tuple[str,int], payload: dict) -> None:
         if not isinstance(peer, tuple) or len(peer) != 2:
             raise ValueError("bad peer")
@@ -42,7 +170,10 @@ def send_to_peer(self, peer: tuple[str,int], payload: dict) -> None:
 
 
     def relay_presence_async(self, pres: dict, exclude=None) -> None:
-        threading.Thread(target=self._relay_presence, args=(pres, exclude), daemon=True).start()
+        try:
+            get_presence_executor().submit(self._relay_presence, pres, exclude)
+        except Exception:
+            pass
 
 
     def mailbox_put(self, addr, item, ttl_s, per_addr_max, global_max):

src/tsarchain/storage/kv.py

@@ -26,6 +26,7 @@ def get_db_path(name: str) -> str:
         "state": CFG.LMDB_STATE_DIR,
         "graffiti": CFG.LMDB_GRAFFITI_DIR,
         "mempool": CFG.LMDB_MEMPOOL_DIR,
+        "chat_prekeys": CFG.LMDB_CHAT_PREKEYS,
 
         # Keys and secrets
         "node_secrets": CFG.LMDB_KEYS_DIR,
@@ -47,11 +48,13 @@ def _init_native_store(name: str = "chain"):
             return _native_store
 
         drive_override = os.getenv("TSAR_STORAGE_DRIVE_TYPE")
+        map_size_init = int(CFG.LMDB_PREKEYS_SIZE_INIT) if name == "chat_prekeys" else int(CFG.LMDB_MAP_SIZE_INIT)
+        map_size_max = int(CFG.LMDB_PREKEYS_SIZE_MAX) if name == "chat_prekeys" else int(CFG.LMDB_MAP_SIZE_MAX)
         store = _native_open_storage(
             "lmdb",
             path,
-            map_size_init=int(CFG.LMDB_MAP_SIZE_INIT),
-            map_size_max=int(CFG.LMDB_MAP_SIZE_MAX),
+            map_size_init=map_size_init,
+            map_size_max=map_size_max,
             pretty_json=False,
             drive_type=drive_override,
         )

src/tsarchain/utils/config.py

@@ -109,10 +109,13 @@
 LMDB_STATE_DIR     = "data/node/state"  # LMDB environment path for state
 LMDB_GRAFFITI_DIR  = "data/node/graffiti"  # LMDB environment path for graffiti
 LMDB_MEMPOOL_DIR   = "data/node/mempool"  # LMDB environment path for mempool
+LMDB_CHAT_PREKEYS  = "data/node/chat_prekeys"  # LMDB environment path for chat prekeys
 
-LMDB_MAP_SIZE_INIT = 4 * 1024 * 1024  # initial LMDB map size (4 MB)
-LMDB_MAP_SIZE_MAX  = 64 * 1024 * 1024 * 1024  # upper LMDB map cap (64 GB)
-KV_ITER_CHUNK      = 512 # number of entries per chunk when iterating prefix scans (LMDB)
+LMDB_MAP_SIZE_INIT     = 4 * 1024 * 1024  # initial LMDB map size (4 MB)
+LMDB_MAP_SIZE_MAX      = 64 * 1024 * 1024 * 1024  # upper LMDB map cap (64 GB)
+LMDB_PREKEYS_SIZE_INIT = 4 * 1024 * 1024  # initial chat prekeys LMDB size (4 MB)
+LMDB_PREKEYS_SIZE_MAX  = 250 * 1024 * 1024  # max chat prekeys LMDB size (250 MB)
+KV_ITER_CHUNK          = 512 # number of entries per chunk when iterating prefix scans (LMDB)
 
 
 # ---- KEYS & SECRETS DATABASE PATHS (LMDB) ----
@@ -527,8 +530,7 @@
 CHAT_OPK_MIN_THRESHOLD     = 5  # minimum one-time pre-keys kept ready
 CHAT_OPK_REFILL_COUNT      = 20  # number of pre-keys generated when refilling
 CHAT_SPK_ROTATE_INTERVAL_S = 24 * 3600  # seconds between signed pre-key rotations
-CHAT_OPK_MAX_STORED        = 200  # hard cap untuk jumlah OPK yang disimpan node per alamat
-CHAT_HISTORY_MAX_PER_PEER  = 200  # maksimum entri riwayat chat per pasangan alamat
+CHAT_HISTORY_MAX_PER_PEER  = 200  # Maximum paired chat history entries (stored on client)
 
 
 # =============================================================================
@@ -594,11 +596,11 @@
 
 # ---- CHAT LOOKUP THROTTLING ----
 CHAT_LOOKUP_RL_IP_BURST       = 20   # lookup pubkey chat per IP
-CHAT_LOOKUP_RL_IP_WINDOW_S    = 10   # jendela waktu limiter lookup pubkey
-CHAT_LOOKUP_RL_BACKOFF_S      = 5    # backoff setelah limiter lookup pubkey kena
-CHAT_LOOKUP_RL_ADDR_BURST     = 10   # limiter lookup pubkey per alamat
-CHAT_LOOKUP_RL_ADDR_WINDOW_S  = 10 # jendela waktu limiter per alamat
-CHAT_LOOKUP_RL_ADDR_BACKOFF_S = 8 # backoff setelah limiter per alamat kena
+CHAT_LOOKUP_RL_IP_WINDOW_S    = 10   # seconds window for pubkey lookup limiter
+CHAT_LOOKUP_RL_BACKOFF_S      = 5    # backoff after pubkey lookup limiter is triggered
+CHAT_LOOKUP_RL_ADDR_BURST     = 10   # pubkey lookup limiter per address
+CHAT_LOOKUP_RL_ADDR_WINDOW_S  = 10   # seconds window for per-address pubkey lookup limiter
+CHAT_LOOKUP_RL_ADDR_BACKOFF_S = 8    # backoff after per-address pubkey lookup limiter is triggered
 
 
 # ---- USER RPC THROTTLING ----

tests/unit/conftest.py

@@ -34,6 +34,7 @@ def isolate_unit_test_storage(request, tmp_path, monkeypatch):
     monkeypatch.setattr(CFG, "LMDB_STATE_DIR", str(mock_data / "node/state"))
     monkeypatch.setattr(CFG, "LMDB_GRAFFITI_DIR", str(mock_data / "node/graffiti"))
     monkeypatch.setattr(CFG, "LMDB_MEMPOOL_DIR", str(mock_data / "node/mempool"))
+    monkeypatch.setattr(CFG, "LMDB_CHAT_PREKEYS", str(mock_data / "node/chat_prekeys"))
 
     # 2. Patch web database and cache paths
     monkeypatch.setattr(CFG, "WEB_DATABASE_PATH", str(mock_data / "web"))