pyrtan / sourcev0.19.0 · main

pyrtan / src / pyrtan / store.py

7,681 bytes · last changed 2026-10-25 in Members: a browser or a phone joins a hub through a mailbox and keeps its blobs; release 0.5.0

1"""Blobs a hub keeps: its own, and the ones it keeps for others.
2
3A *blob* is a JSON value under a key (``directory``, ``gaps``,
4``traces/2026-10-25``), with its SHA-256, size and time. A hub keeps its
5own blobs so it can hand them to the members and hubs that follow it, and
6keeps blobs *for* the hubs and members it shares a secret with, so that
7when their box dies the blobs are still somewhere. Nothing secret is ever
8a blob: the state file, with the pair secrets, never goes through here.
9
10On disk (``--store DIR``) the layout is ``own/<key>.json`` and
11``kept/<actor-hash>/<key>.json``; without a directory the store lives in
12memory and is lost with the process, which is enough to try the loop.
13Limits are fixed and small on purpose: a blob is at most 256 KB, an
14actor may leave at most 200 keys and 64 MB here. A keeper that is asked
15for more says so and keeps nothing extra.
16"""
17
18from __future__ import annotations
19
20import hashlib
21import json
22import re
23import threading
24import time
25from pathlib import Path
26from typing import Any
27
28MAX_BLOB = 256 * 1024
29MAX_KEYS_PER_ACTOR = 200
30MAX_BYTES_PER_ACTOR = 64 * 1024 * 1024
31KEY = re.compile(r"^[a-z0-9][a-z0-9._/-]{0,99}$")
32
33
34class StoreError(ValueError):
35 pass
36
37
38def canonical(data: Any) -> bytes:
39 return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
40
41
42def digest(data: Any) -> tuple[str, int]:
43 raw = canonical(data)
44 return hashlib.sha256(raw).hexdigest(), len(raw)
45
46
47def check_key(key: Any) -> str:
48 if not isinstance(key, str) or not KEY.match(key) or ".." in key:
49 raise StoreError(f"{key!r} is not a key: lowercase letters, digits, dot, dash, underscore, slash; at most 100 characters")
50 return key
51
52
53def _safe(key: str) -> str:
54 return key.replace("/", "__")
55
56
57def _actor_dir(actor: str) -> str:
58 return hashlib.sha256(actor.rstrip("/").encode("utf-8")).hexdigest()[:16]
59
60
61class BlobStore:
62 def __init__(self, root: str | Path | None = None) -> None:
63 self.root = Path(root) if root else None
64 self._own: dict[str, dict[str, Any]] = {} # key -> blob
65 self._kept: dict[str, dict[str, dict[str, Any]]] = {} # actor -> key -> blob
66 self._lock = threading.RLock()
67 if self.root is not None:
68 self._load()
69
70 # ------------------------------------------------------------ disk
71
72 def _load(self) -> None:
73 assert self.root is not None
74 for path in sorted((self.root / "own").glob("*.json")) if (self.root / "own").is_dir() else []:
75 blob = self._read(path)
76 if blob:
77 self._own[blob["key"]] = blob
78 kept = self.root / "kept"
79 for actor_dir in sorted(kept.iterdir()) if kept.is_dir() else []:
80 for path in sorted(actor_dir.glob("*.json")):
81 blob = self._read(path)
82 if blob and isinstance(blob.get("actor"), str):
83 self._kept.setdefault(blob["actor"], {})[blob["key"]] = blob
84
85 @staticmethod
86 def _read(path: Path) -> dict[str, Any] | None:
87 try:
88 blob = json.loads(path.read_text(encoding="utf-8"))
89 except (OSError, json.JSONDecodeError):
90 return None
91 return blob if isinstance(blob, dict) and isinstance(blob.get("key"), str) else None
92
93 def _write(self, path: Path, blob: dict[str, Any]) -> None:
94 if self.root is None:
95 return
96 path.parent.mkdir(parents=True, exist_ok=True)
97 tmp = path.with_suffix(".tmp")
98 tmp.write_text(json.dumps(blob, ensure_ascii=False) + "\n", encoding="utf-8")
99 tmp.replace(path)
100
101 def _remove(self, path: Path) -> None:
102 if self.root is not None:
103 try:
104 path.unlink()
105 except OSError:
106 pass
107
108 # ------------------------------------------------------------ own
109
110 @staticmethod
111 def meta(blob: dict[str, Any]) -> dict[str, Any]:
112 return {k: blob[k] for k in ("key", "sha256", "size", "updated") if k in blob}
113
114 def put_own(self, key: str, data: Any) -> dict[str, Any]:
115 key = check_key(key)
116 sha, size = digest(data)
117 if size > MAX_BLOB:
118 raise StoreError(f"{key}: {size} bytes; a blob is at most {MAX_BLOB}")
119 with self._lock:
120 current = self._own.get(key)
121 if current is not None and current["sha256"] == sha:
122 return self.meta(current)
123 blob = {"key": key, "sha256": sha, "size": size, "updated": time.time(), "data": data}
124 self._own[key] = blob
125 if self.root is not None:
126 self._write(self.root / "own" / f"{_safe(key)}.json", blob)
127 return self.meta(blob)
128
129 def get_own(self, key: str) -> dict[str, Any] | None:
130 with self._lock:
131 blob = self._own.get(key)
132 return dict(blob) if blob else None
133
134 def list_own(self) -> list[dict[str, Any]]:
135 with self._lock:
136 return [self.meta(b) for _, b in sorted(self._own.items())]
137
138 # ------------------------------------------------------------ kept for others
139
140 def put_kept(self, actor: str, key: str, data: Any, *, sha256: str | None = None, updated: float | None = None) -> dict[str, Any]:
141 key = check_key(key)
142 actor = actor.rstrip("/")
143 sha, size = digest(data)
144 if sha256 is not None and sha256 != sha:
145 raise StoreError(f"{key}: the data does not match its sha256")
146 if size > MAX_BLOB:
147 raise StoreError(f"{key}: {size} bytes; a blob is at most {MAX_BLOB}")
148 with self._lock:
149 mine = self._kept.setdefault(actor, {})
150 others = sum(b["size"] for k, b in mine.items() if k != key)
151 if key not in mine and len(mine) >= MAX_KEYS_PER_ACTOR:
152 raise StoreError(f"{actor} already has {MAX_KEYS_PER_ACTOR} keys here")
153 if others + size > MAX_BYTES_PER_ACTOR:
154 raise StoreError(f"{actor} would have more than {MAX_BYTES_PER_ACTOR} bytes here")
155 blob = {"actor": actor, "key": key, "sha256": sha, "size": size, "updated": float(updated or time.time()), "kept_at": time.time(), "data": data}
156 mine[key] = blob
157 if self.root is not None:
158 self._write(self.root / "kept" / _actor_dir(actor) / f"{_safe(key)}.json", blob)
159 return {**self.meta(blob), "actor": actor}
160
161 def get_kept(self, actor: str, key: str) -> dict[str, Any] | None:
162 with self._lock:
163 blob = self._kept.get(actor.rstrip("/"), {}).get(key)
164 return dict(blob) if blob else None
165
166 def list_kept(self, actor: str | None = None) -> list[dict[str, Any]]:
167 with self._lock:
168 actors = [actor.rstrip("/")] if actor else sorted(self._kept)
169 return [{**self.meta(b), "actor": a} for a in actors for _, b in sorted(self._kept.get(a, {}).items())]
170
171 def drop_kept(self, actor: str) -> int:
172 actor = actor.rstrip("/")
173 with self._lock:
174 mine = self._kept.pop(actor, {})
175 for key in mine:
176 self._remove((self.root or Path(".")) / "kept" / _actor_dir(actor) / f"{_safe(key)}.json") if self.root else None
177 return len(mine)
178
179 def usage(self) -> dict[str, Any]:
180 with self._lock:
181 return {
182 "own": len(self._own),
183 "kept": sum(len(m) for m in self._kept.values()),
184 "kept_bytes": sum(b["size"] for m in self._kept.values() for b in m.values()),
185 "actors": len(self._kept),
186 "root": str(self.root) if self.root else None,
187 }