File size: 2,772 Bytes
bdfb884 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """
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
# Gear table: 256 fixed 64-bit values (deterministic).
_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] = {} # sha256 -> chunk 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()),
}
|