Spaces:
Running
Running
File size: 5,298 Bytes
a6a5d8e | 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 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """KipuPool — the shared, content-addressed receipt-cell substrate.
Every organ reads/writes ReceiptCells here. Each write:
1. computes/verifies the content address (cid),
2. persists the cell (LMDB if available, else a JSON-file store — same API),
3. optionally stores a Reed-Solomon shard manifest for durability,
4. publishes a ("write", cell) event on the in-process EventBus.
Each read publishes a ("read", cid) event and emits a read-receipt cell, so there is no
silent access. This is a Linda-style tuple space + event sourcing + content addressing.
HONEST: durability is Reed-Solomon erasure coding, not "holographic QEC".
"""
from __future__ import annotations
import json
import os
import threading
from pathlib import Path
from typing import Iterable, Optional
from .cell import ReceiptCell
from .coding import decode_shards, encode_cell
from .events import EventBus
class _Store:
"""Key-value persistence: LMDB backend if installed, else JSON-file fallback."""
def __init__(self, path: str):
self.path = path
self.backend = "json"
self._lmdb = None
self._lock = threading.RLock()
try:
import lmdb # type: ignore
os.makedirs(path, exist_ok=True)
self._lmdb = lmdb.open(path, map_size=256 * 1024 * 1024)
self.backend = "lmdb"
except Exception:
self._file = Path(path)
self._file.parent.mkdir(parents=True, exist_ok=True)
if not self._file.exists():
self._file.write_text("{}")
def put(self, key: str, value: bytes) -> None:
with self._lock:
if self.backend == "lmdb":
with self._lmdb.begin(write=True) as txn:
txn.put(key.encode(), value)
else:
d = json.loads(self._file.read_text())
d[key] = value.decode("utf-8")
self._file.write_text(json.dumps(d))
def get(self, key: str) -> Optional[bytes]:
with self._lock:
if self.backend == "lmdb":
with self._lmdb.begin() as txn:
v = txn.get(key.encode())
return bytes(v) if v is not None else None
d = json.loads(self._file.read_text())
v = d.get(key)
return v.encode("utf-8") if v is not None else None
def keys(self) -> Iterable[str]:
with self._lock:
if self.backend == "lmdb":
with self._lmdb.begin() as txn:
return [k.decode() for k, _ in txn.cursor()]
return list(json.loads(self._file.read_text()).keys())
class KipuPool:
"""The shared receipt-cell substrate."""
def __init__(self, path: str = "/tmp/kipu", durability: bool = True,
rs_n: int = 10, rs_k: int = 6):
self.bus = EventBus()
self._cells = _Store(os.path.join(path, "cells"))
self._shards = _Store(os.path.join(path, "shards"))
self.durability = durability
self.rs_n, self.rs_k = rs_n, rs_k
@property
def store_backend(self) -> str:
return self._cells.backend
def write(self, cell: ReceiptCell) -> str:
"""Persist a cell, optionally RS-encode it, publish a write event. Returns cid."""
if not cell.verify():
raise ValueError("ReceiptCell failed content-address verification")
cb = cell.to_bytes()
self._cells.put(cell.cid, cb)
if self.durability:
manifest = encode_cell(cb, self.rs_n, self.rs_k)
self._shards.put(cell.cid, json.dumps(manifest).encode())
self.bus.publish("write", cell)
self.bus.publish(f"organ:{cell.organ}", cell)
return cell.cid
def read(self, cid: str, reader: str = "anon", emit_receipt: bool = True) -> Optional[ReceiptCell]:
"""Read a cell by cid. Publishes a read event and (optionally) a read-receipt."""
raw = self._cells.get(cid)
cell = ReceiptCell.from_bytes(raw) if raw is not None else None
self.bus.publish("read", {"cid": cid, "reader": reader, "hit": cell is not None})
if emit_receipt and cell is not None:
rr = ReceiptCell(organ=reader, kind="read_receipt",
payload={"read_cid": cid}, parents=(cid,))
self._cells.put(rr.cid, rr.to_bytes())
return cell
def recover(self, cid: str, drop: Optional[list[int]] = None) -> Optional[ReceiptCell]:
"""Recover a cell from its Reed-Solomon shards, optionally simulating lost shards."""
raw = self._shards.get(cid)
if raw is None:
return None
manifest = json.loads(raw.decode())
if drop:
shards = list(manifest["shards"])
for i in drop:
shards[i] = None
manifest = {**manifest, "shards": shards}
cb = decode_shards(manifest)
return ReceiptCell.from_bytes(cb)
def all_cids(self) -> list[str]:
return list(self._cells.keys())
def stats(self) -> dict:
return {
"store_backend": self._cells.backend,
"cells": len(list(self._cells.keys())),
"durability": self.durability,
"rs_code": f"RS({self.rs_n},{self.rs_k})",
}
|