diff --git a/cosmic_ledger/delta.py b/cosmic_ledger/delta.py index 8eb2b65..635454c 100644 --- a/cosmic_ledger/delta.py +++ b/cosmic_ledger/delta.py @@ -2,6 +2,7 @@ import hashlib import json from typing import List from .crypto import Signer +import math def _hash_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() @@ -115,6 +116,57 @@ def verify_inclusion(leaf_digest: str, merkle_path: List[str], root: str, index: return node.hex() == (root or "") + +class BloomFilter: + """A minimal Bloom filter implementation used for compact existence indexes. + + This is intentionally small and dependency-free: it stores bits in a bytearray + and uses multiple SHA-256-based hash functions with different salts. + """ + def __init__(self, m: int = 1024, k: int = 3): + if m <= 0: + raise ValueError("m (number of bits) must be > 0") + if k <= 0: + raise ValueError("k (number of hash functions) must be > 0") + self.m = int(m) + self.k = int(k) + self._bits = bytearray((self.m + 7) // 8) + + def _hashes(self, item: str): + if isinstance(item, str): + item = item.encode("utf-8") + for i in range(self.k): + h = hashlib.sha256() + # include the index as a single byte salt + h.update(i.to_bytes(1, "little")) + h.update(item) + yield int.from_bytes(h.digest(), "big") % self.m + + def add(self, item: str) -> None: + for pos in self._hashes(item): + byte_index = pos // 8 + bit_index = pos % 8 + self._bits[byte_index] |= (1 << bit_index) + + def __contains__(self, item: str) -> bool: + for pos in self._hashes(item): + byte_index = pos // 8 + bit_index = pos % 8 + if not (self._bits[byte_index] & (1 << bit_index)): + return False + return True + + def to_dict(self) -> dict: + return {"m": self.m, "k": self.k, "bits": self._bits.hex()} + + @staticmethod + def from_dict(d: dict) -> "BloomFilter": + bf = BloomFilter(m=d.get("m", 1024), k=d.get("k", 3)) + bits_hex = d.get("bits", "") + if bits_hex: + bf._bits = bytearray.fromhex(bits_hex) + return bf + class DeltaLog: def __init__(self): self.entries = [] # each entry is a dict with digest and payload @@ -225,14 +277,26 @@ class DeltaLog: # nothing to archive cp_digests = [] cp_root = "" + cp_bloom = None else: cp_digests = [e["digest"] for e in self.entries[:prune_before_index]] cp_root = merkle_root(cp_digests) + # Build a compact Bloom filter summarizing archived digests to + # allow peers to quickly test probable presence before requesting + # proofs or shards. Choose conservative defaults for m and k for + # typical small checkpoints; these can be tuned per mission. + m = max(1024, len(cp_digests) * 16) # bits + k = 4 + bf = BloomFilter(m=m, k=k) + for d in cp_digests: + bf.add(d) + cp_bloom = bf.to_dict() checkpoint = { "root": cp_root, "index": prune_before_index, "digests": cp_digests, + "bloom": cp_bloom, } # store checkpoint self.checkpoints.append(checkpoint) diff --git a/tests/test_bloom.py b/tests/test_bloom.py new file mode 100644 index 0000000..3f9c33b --- /dev/null +++ b/tests/test_bloom.py @@ -0,0 +1,33 @@ +import sys +import pathlib + +# Ensure repository root is on sys.path so tests can import the package +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from cosmic_ledger.delta import DeltaLog, BloomFilter + + +def test_checkpoint_includes_bloom_and_membership(): + dl = DeltaLog() + # add several payloads + for i in range(8): + dl.add_entry({"v": i}) + + # create checkpoint archiving first 5 entries + cp = dl.checkpoint(prune_before_index=5) + + assert "bloom" in cp + assert cp["bloom"] is not None + + # load bloom and check that all archived digests are (probably) present + bf = BloomFilter.from_dict(cp["bloom"]) + archived = cp["digests"] + # Bloom filters may have false positives but must not have false negatives + for d in archived: + assert d in bf + + # non-archived digest may or may not be in bloom (allow either) + remaining = [e["digest"] for e in dl.entries] + if remaining: + # ensure at least one remaining digest is not erroneously marked as missing + _ = remaining[0] # presence not asserted