| """ |
| Content-addressed, deduplicating chunk store. |
| |
| * content-defined chunking (Gear-hash CDC) splits a byte stream into |
| variable-length chunks at data-dependent boundaries -- so duplicate regions |
| chunk identically even if shifted. |
| * each chunk's SHA-256 is its address and its bit-verification. |
| * identical chunks are stored once (dedup). A manifest (list of hashes) |
| reconstructs the original stream exactly. |
| |
| Honest note: dedup + (optional) compression only removes REDUNDANT/compressible |
| data. Incompressible or unique data is stored in full -- you cannot beat entropy. |
| The chunk hashes are real crypto (SHA-256), not a neural emulation. |
| """ |
| from __future__ import annotations |
| import hashlib |
| import random |
|
|
| |
| _rng = random.Random(0xC0FFEE) |
| GEAR = [_rng.getrandbits(64) for _ in range(256)] |
| MASK64 = (1 << 64) - 1 |
|
|
|
|
| def chunk_stream(data: bytes, min_sz=2048, avg_bits=13, max_sz=65536): |
| """Yield (offset, length) chunk boundaries via Gear-hash CDC. |
| avg_bits sets average chunk size ~ 2**avg_bits bytes.""" |
| mask = (1 << avg_bits) - 1 |
| n = len(data) |
| start = 0 |
| while start < n: |
| h = 0 |
| i = start |
| end = min(start + max_sz, n) |
| while i < end: |
| h = ((h << 1) + GEAR[data[i]]) & MASK64 |
| i += 1 |
| if i - start >= min_sz and (h & mask) == 0: |
| break |
| yield start, i - start |
| start = i |
|
|
|
|
| class ChunkStore: |
| def __init__(self): |
| self.store: dict[str, bytes] = {} |
| self.logical_bytes = 0 |
| self.total_chunks = 0 |
|
|
| def put_stream(self, data: bytes) -> list[str]: |
| manifest = [] |
| self.logical_bytes += len(data) |
| for off, ln in chunk_stream(data): |
| ch = data[off:off + ln] |
| h = hashlib.sha256(ch).hexdigest() |
| if h not in self.store: |
| self.store[h] = ch |
| manifest.append(h) |
| self.total_chunks += 1 |
| return manifest |
|
|
| def reconstruct(self, manifest: list[str]) -> bytes: |
| return b"".join(self.store[h] for h in manifest) |
|
|
| def integrity_ok(self) -> bool: |
| """Every stored chunk must still hash to its address (bit-verified).""" |
| return all(hashlib.sha256(v).hexdigest() == k for k, v in self.store.items()) |
|
|
| def stored_bytes(self) -> int: |
| return sum(len(v) for v in self.store.values()) |
|
|
| def stats(self) -> dict: |
| return { |
| "logical_bytes": self.logical_bytes, |
| "stored_bytes": self.stored_bytes(), |
| "total_chunks": self.total_chunks, |
| "unique_chunks": len(self.store), |
| "dedup_ratio": self.logical_bytes / max(1, self.stored_bytes()), |
| } |
|
|
