feat(backup): add media offload data layer · solpbc.org/solstone@1ce01d8

+1

AGENTS.md

+1

scripts/check_journal_io_access.py

View file

··· 125 125 "solstone/think/identity.py", 126 126 "solstone/think/journal_config.py", 127 127 "solstone/think/log_retention.py", 128 + "solstone/think/offload_ledger.py", 128 129 # Sole writer of content-free bundled-local inference telemetry. 129 130 "solstone/think/providers/local_admission.py", 130 131 "solstone/think/schedule_config.py",

+106

solstone/think/backup/state.py

View file

··· 39 39 "weekly": 4, 40 40 "monthly": 12, 41 41 }, 42 + "offload": { 43 + "enabled": False, 44 + "budget_bytes": None, 45 + "floor_bytes": None, 46 + }, 42 47 "schedule": { 43 48 "every": "daily", 44 49 "enabled": False, ··· 54 59 "status": None, 55 60 "error_reason": None, 56 61 }, 62 + # Offload records use "reason" instead of "error_reason" because skipped 63 + # and stalled are expected outcomes, not errors. 64 + "last_offload": { 65 + "time": None, 66 + "status": None, 67 + "reason": None, 68 + }, 69 + "last_verification": { 70 + "time": None, 71 + "status": None, 72 + "reason": None, 73 + }, 57 74 } 75 + OFFLOAD_KEYS = ("enabled", "budget_bytes", "floor_bytes") 76 + OFFLOAD_STATUSES = ("ok", "skipped", "stalled", "error") 58 77 RETENTION_KEYS = ("hourly", "daily", "weekly", "monthly") 78 + VERIFICATION_STATUSES = ("ok", "skipped", "error") 59 79 60 80 61 81 @dataclass(frozen=True) ··· 200 220 write_journal_config(config) 201 221 202 222 223 + def set_offload(offload: dict[str, Any]) -> None: 224 + if not isinstance(offload, dict): 225 + raise ValueError("backup offload must be a JSON object") 226 + if set(offload) != set(OFFLOAD_KEYS): 227 + raise ValueError( 228 + "backup offload must include enabled, budget_bytes, floor_bytes" 229 + ) 230 + if not isinstance(offload["enabled"], bool): 231 + raise ValueError("backup offload enabled must be a boolean") 232 + for key in ("budget_bytes", "floor_bytes"): 233 + value = offload[key] 234 + if value is not None and (type(value) is not int or value <= 0): 235 + raise ValueError( 236 + "backup offload byte values must be positive integers or null" 237 + ) 238 + 239 + with hold_config_lock(): 240 + config = read_journal_config() 241 + backup = _writable_backup_section(config) 242 + backup["offload"] = { 243 + "enabled": offload["enabled"], 244 + "budget_bytes": offload["budget_bytes"], 245 + "floor_bytes": offload["floor_bytes"], 246 + } 247 + write_journal_config(config) 248 + 249 + 203 250 def set_recovery_key(recovery_key: str) -> None: 204 251 with hold_config_lock(): 205 252 config = read_journal_config() ··· 251 298 write_journal_config(config) 252 299 253 300 301 + def record_offload_result( 302 + *, 303 + status: str, 304 + time: int | None, 305 + reason: str | None = None, 306 + ) -> None: 307 + """Record the last media-offload run. 308 + 309 + Convention: reason is None on ok. This layer only enforces the closed 310 + status vocabulary; callers own richer run-state validation. 311 + """ 312 + if status not in OFFLOAD_STATUSES: 313 + raise ValueError("backup offload status must be ok, skipped, stalled, or error") 314 + 315 + with hold_config_lock(): 316 + config = read_journal_config() 317 + backup = _writable_backup_section(config) 318 + backup["last_offload"] = { 319 + "time": time, 320 + "status": status, 321 + "reason": reason, 322 + } 323 + write_journal_config(config) 324 + 325 + 326 + def record_verification_result( 327 + *, 328 + status: str, 329 + time: int | None, 330 + reason: str | None = None, 331 + ) -> None: 332 + """Record the last media-offload verification run. 333 + 334 + Convention: reason is None on ok. This layer only enforces the closed 335 + status vocabulary; callers own richer run-state validation. 336 + """ 337 + if status not in VERIFICATION_STATUSES: 338 + raise ValueError("backup verification status must be ok, skipped, or error") 339 + 340 + with hold_config_lock(): 341 + config = read_journal_config() 342 + backup = _writable_backup_section(config) 343 + backup["last_verification"] = { 344 + "time": time, 345 + "status": status, 346 + "reason": reason, 347 + } 348 + write_journal_config(config) 349 + 350 + 254 351 def status_view() -> dict[str, Any]: 255 352 config = get_backup_config() 256 353 destination = config["destination"] ··· 275 372 "recovery_key_set": config["recovery_key"] is not None, 276 373 "recovery_key_confirmed": bool(config["confirmed_recovery_key"]), 277 374 "retention": config["retention"], 375 + "offload": config["offload"], 278 376 "schedule": config["schedule"], 279 377 "last_backup": config["last_backup"], 280 378 "last_prune": config["last_prune"], 379 + "last_offload": config["last_offload"], 380 + "last_verification": config["last_verification"], 281 381 "hosted": hosted, 282 382 } 283 383 ··· 285 385 __all__ = [ 286 386 "BACKUP_DEFAULTS", 287 387 "BackupKeys", 388 + "OFFLOAD_KEYS", 389 + "OFFLOAD_STATUSES", 390 + "VERIFICATION_STATUSES", 288 391 "clear_backup_config", 289 392 "generate_and_store_keys", 290 393 "get_backup_config", 291 394 "get_destination", 292 395 "get_keys", 293 396 "record_backup_result", 397 + "record_offload_result", 294 398 "record_prune_result", 399 + "record_verification_result", 295 400 "set_destination", 296 401 "set_enabled", 297 402 "set_mode", 403 + "set_offload", 298 404 "set_recovery_key", 299 405 "set_recovery_key_confirmed", 300 406 "set_retention",

+496

solstone/think/offload_ledger.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Durable media-offload ledger. 5 + 6 + This ledger is JSONL, not sqlite, because backup excludes ``*.sqlite*`` while 7 + ``health/`` is included. Losing the journal must not lose the map to the 8 + owner's only backed-up copy of their media. It is fsync'd even though the 9 + nearby pruning audit writer is not: pruning audits are post-hoc records, while 10 + this ledger is a pre-delete witness for a later path that deletes local media 11 + immediately after append returns. 12 + 13 + Per-file records intentionally use ``name`` and ``bytes`` to match the existing 14 + raw-media audit shape, but use ``sha256`` instead of ``hash`` because this 15 + durable schema rides encrypted backups and may be read years later by a restore 16 + flow. A future delete path should compute each digest once, feed ``sha256`` to 17 + this ledger, and remap only that key to the audit writer's ``hash`` at the call 18 + site; it must not hash twice. 19 + 20 + Restore events carry only the identity spine and time. Repeating old file or 21 + snapshot facts on restore would invite ambiguous folded reads; restore simply 22 + invalidates the current offload state for a segment. 23 + 24 + Fold order is append order. Timestamps are informational, so a clock step 25 + backward must not change the winning event. Read degradation is also deliberate: 26 + unlike the repo's usual fail-loudly rule, a crashing read could let a future 27 + teardown gate lose the owner's only media copy. A degraded zero is not a clean 28 + zero; teardown gates must check ``degraded is False`` before trusting zero 29 + offloaded bytes or segments. 30 + """ 31 + 32 + from __future__ import annotations 33 + 34 + import json 35 + import logging 36 + import re 37 + import time as time_module 38 + from dataclasses import dataclass 39 + from pathlib import Path 40 + from typing import Any, Literal 41 + 42 + from solstone.think.journal_io import append_jsonl 43 + from solstone.think.utils import ( 44 + DATE_RE, 45 + DEFAULT_STREAM, 46 + STREAM_RE, 47 + get_journal, 48 + segment_key, 49 + ) 50 + 51 + logger = logging.getLogger(__name__) 52 + 53 + EVENT_OFFLOAD = "offload" 54 + EVENT_RESTORE = "restore" 55 + SHA256_RE = re.compile(r"[0-9a-f]{64}") 56 + 57 + 58 + @dataclass(frozen=True) 59 + class OffloadFile: 60 + name: str 61 + bytes: int 62 + sha256: str 63 + 64 + 65 + @dataclass(frozen=True) 66 + class SegmentOffloadSummary: 67 + day: str 68 + stream: str 69 + segment: str 70 + currently_offloaded: bool 71 + snapshot_id: str | None 72 + files: tuple[OffloadFile, ...] 73 + offloaded_bytes: int 74 + offloaded_file_count: int 75 + skipped_records: int 76 + unreadable_ledgers: tuple[str, ...] 77 + 78 + @property 79 + def degraded(self) -> bool: 80 + return bool(self.unreadable_ledgers) 81 + 82 + 83 + @dataclass(frozen=True) 84 + class DayOffloadSummary: 85 + day: str 86 + segments: tuple[SegmentOffloadSummary, ...] 87 + offloaded_bytes: int 88 + offloaded_file_count: int 89 + offloaded_segments: int 90 + skipped_records: int 91 + unreadable_ledgers: tuple[str, ...] 92 + 93 + @property 94 + def degraded(self) -> bool: 95 + return bool(self.unreadable_ledgers) 96 + 97 + 98 + @dataclass(frozen=True) 99 + class JournalOffloadSummary: 100 + days: tuple[DayOffloadSummary, ...] 101 + offloaded_bytes: int 102 + offloaded_file_count: int 103 + offloaded_segments: int 104 + offloaded_days: int 105 + skipped_records: int 106 + unreadable_ledgers: tuple[str, ...] 107 + 108 + @property 109 + def degraded(self) -> bool: 110 + return bool(self.unreadable_ledgers) 111 + 112 + 113 + @dataclass(frozen=True) 114 + class _LedgerEvent: 115 + event_kind: Literal["offload", "restore"] 116 + time: int 117 + day: str 118 + stream: str 119 + segment: str 120 + snapshot_id: str | None 121 + files: tuple[OffloadFile, ...] 122 + 123 + 124 + @dataclass(frozen=True) 125 + class _LedgerRead: 126 + events: tuple[_LedgerEvent, ...] 127 + skipped_records: int 128 + unreadable_ledgers: tuple[str, ...] 129 + 130 + 131 + def append_offload_event( 132 + *, 133 + day: str, 134 + stream: str, 135 + segment: str, 136 + snapshot_id: str, 137 + files: tuple[OffloadFile, ...] | list[OffloadFile], 138 + time: int | None = None, 139 + ) -> None: 140 + event_time = _write_event_time(time) 141 + _validate_identity(day, stream, segment) 142 + if not isinstance(snapshot_id, str) or not snapshot_id: 143 + raise ValueError("snapshot_id must be a non-empty string") 144 + if not files: 145 + raise ValueError("files must not be empty") 146 + validated_files = tuple(_validate_file_record(file) for file in files) 147 + 148 + append_jsonl( 149 + _ledger_path(day), 150 + { 151 + "event_kind": EVENT_OFFLOAD, 152 + "time": event_time, 153 + "day": day, 154 + "stream": stream, 155 + "segment": segment, 156 + "snapshot_id": snapshot_id, 157 + "files": [_file_to_record(file) for file in validated_files], 158 + }, 159 + ) 160 + 161 + 162 + def append_restore_event( 163 + *, 164 + day: str, 165 + stream: str, 166 + segment: str, 167 + time: int | None = None, 168 + ) -> None: 169 + event_time = _write_event_time(time) 170 + _validate_identity(day, stream, segment) 171 + 172 + append_jsonl( 173 + _ledger_path(day), 174 + { 175 + "event_kind": EVENT_RESTORE, 176 + "time": event_time, 177 + "day": day, 178 + "stream": stream, 179 + "segment": segment, 180 + }, 181 + ) 182 + 183 + 184 + def summarize_segment(day: str, stream: str, segment: str) -> SegmentOffloadSummary: 185 + _validate_identity(day, stream, segment) 186 + read = _read_ledger_day(day) 187 + states = _fold_events(read.events) 188 + current = states.get((day, stream, segment)) 189 + return _segment_summary( 190 + day, 191 + stream, 192 + segment, 193 + current, 194 + skipped_records=read.skipped_records, 195 + unreadable_ledgers=read.unreadable_ledgers, 196 + ) 197 + 198 + 199 + def summarize_day(day: str) -> DayOffloadSummary: 200 + if not DATE_RE.fullmatch(day): 201 + raise ValueError("day must be in YYYYMMDD format") 202 + read = _read_ledger_day(day) 203 + states = _fold_events(read.events) 204 + segments = tuple( 205 + _segment_summary( 206 + key_day, 207 + stream, 208 + segment, 209 + current, 210 + skipped_records=0, 211 + unreadable_ledgers=(), 212 + ) 213 + for (key_day, stream, segment), current in sorted(states.items()) 214 + if key_day == day 215 + ) 216 + return _day_summary( 217 + day, 218 + segments, 219 + skipped_records=read.skipped_records, 220 + unreadable_ledgers=read.unreadable_ledgers, 221 + ) 222 + 223 + 224 + def summarize_journal() -> JournalOffloadSummary: 225 + ledger_dir = _ledger_dir() 226 + if not ledger_dir.is_dir(): 227 + return JournalOffloadSummary( 228 + days=(), 229 + offloaded_bytes=0, 230 + offloaded_file_count=0, 231 + offloaded_segments=0, 232 + offloaded_days=0, 233 + skipped_records=0, 234 + unreadable_ledgers=(), 235 + ) 236 + 237 + days = tuple( 238 + summarize_day(path.stem) 239 + for path in sorted(ledger_dir.glob("*.jsonl")) 240 + if DATE_RE.fullmatch(path.stem) 241 + ) 242 + unreadable_ledgers = tuple( 243 + ledger for day in days for ledger in day.unreadable_ledgers 244 + ) 245 + return JournalOffloadSummary( 246 + days=days, 247 + offloaded_bytes=sum(day.offloaded_bytes for day in days), 248 + offloaded_file_count=sum(day.offloaded_file_count for day in days), 249 + offloaded_segments=sum(day.offloaded_segments for day in days), 250 + offloaded_days=sum(1 for day in days if day.offloaded_segments > 0), 251 + skipped_records=sum(day.skipped_records for day in days), 252 + unreadable_ledgers=unreadable_ledgers, 253 + ) 254 + 255 + 256 + def ledger_path_for_day(day: str) -> Path: 257 + if not DATE_RE.fullmatch(day): 258 + raise ValueError("day must be in YYYYMMDD format") 259 + return _ledger_path(day) 260 + 261 + 262 + def _ledger_dir() -> Path: 263 + return Path(get_journal()) / "health" / "offload" 264 + 265 + 266 + def _ledger_path(day: str) -> Path: 267 + return _ledger_dir() / f"{day}.jsonl" 268 + 269 + 270 + def _write_event_time(value: int | None) -> int: 271 + if value is None: 272 + return int(time_module.time()) 273 + return _read_event_time(value) 274 + 275 + 276 + def _read_event_time(value: Any) -> int: 277 + if type(value) is not int or value < 0: 278 + raise ValueError("time must be a non-negative integer epoch second") 279 + return value 280 + 281 + 282 + def _validate_identity(day: str, stream: str, segment: str) -> None: 283 + if not isinstance(day, str) or not DATE_RE.fullmatch(day): 284 + raise ValueError("day must be in YYYYMMDD format") 285 + if not isinstance(stream, str) or not ( 286 + stream == DEFAULT_STREAM or STREAM_RE.fullmatch(stream) 287 + ): 288 + raise ValueError("stream must be a valid stream name") 289 + if not isinstance(segment, str) or segment_key(segment) != segment: 290 + raise ValueError("segment must be an exact segment key") 291 + 292 + 293 + def _validate_file_record(file: OffloadFile) -> OffloadFile: 294 + if not isinstance(file, OffloadFile): 295 + raise ValueError("files must contain OffloadFile records") 296 + _validate_file_fields(file.name, file.bytes, file.sha256) 297 + return file 298 + 299 + 300 + def _validate_file_fields(name: Any, size: Any, sha256: Any) -> None: 301 + if not isinstance(name, str) or not name or "/" in name or name in {".", ".."}: 302 + raise ValueError("file name must be a segment-local basename") 303 + if type(size) is not int or size < 0: 304 + raise ValueError("file bytes must be a non-negative integer") 305 + if not isinstance(sha256, str) or SHA256_RE.fullmatch(sha256) is None: 306 + raise ValueError("file sha256 must be a lowercase 64-character hex digest") 307 + 308 + 309 + def _file_to_record(file: OffloadFile) -> dict[str, Any]: 310 + return {"name": file.name, "bytes": file.bytes, "sha256": file.sha256} 311 + 312 + 313 + def _read_ledger_day(day: str) -> _LedgerRead: 314 + return _read_ledger_file(_ledger_path(day), expected_day=day) 315 + 316 + 317 + def _read_ledger_file(path: Path, *, expected_day: str | None) -> _LedgerRead: 318 + if not path.exists(): 319 + return _LedgerRead(events=(), skipped_records=0, unreadable_ledgers=()) 320 + 321 + try: 322 + raw = path.read_text(encoding="utf-8") 323 + except (OSError, UnicodeDecodeError) as exc: 324 + logger.warning("offload ledger read degraded for %s: %s", path, exc) 325 + return _LedgerRead( 326 + events=(), skipped_records=0, unreadable_ledgers=(str(path),) 327 + ) 328 + 329 + events: list[_LedgerEvent] = [] 330 + skipped_records = 0 331 + for lineno, line in enumerate(raw.splitlines(), start=1): 332 + if not line.strip(): 333 + continue 334 + try: 335 + record = json.loads(line) 336 + event = _parse_event(record) 337 + if expected_day is not None and event.day != expected_day: 338 + raise ValueError("record day does not match ledger file day") 339 + except json.JSONDecodeError: 340 + skipped_records += 1 341 + logger.warning( 342 + "skipping malformed offload ledger line %d in %s", lineno, path 343 + ) 344 + continue 345 + except (TypeError, ValueError) as exc: 346 + skipped_records += 1 347 + logger.warning( 348 + "skipping invalid offload ledger record line %d in %s: %s", 349 + lineno, 350 + path, 351 + exc, 352 + ) 353 + continue 354 + events.append(event) 355 + return _LedgerRead( 356 + events=tuple(events), skipped_records=skipped_records, unreadable_ledgers=() 357 + ) 358 + 359 + 360 + def _parse_event(record: Any) -> _LedgerEvent: 361 + if not isinstance(record, dict): 362 + raise ValueError("record must be a JSON object") 363 + event_kind = record.get("event_kind") 364 + if event_kind == EVENT_OFFLOAD: 365 + expected = { 366 + "event_kind", 367 + "time", 368 + "day", 369 + "stream", 370 + "segment", 371 + "snapshot_id", 372 + "files", 373 + } 374 + if set(record) != expected: 375 + raise ValueError("offload record has unexpected fields") 376 + _validate_identity(record["day"], record["stream"], record["segment"]) 377 + event_time = _read_event_time(record["time"]) 378 + snapshot_id = record["snapshot_id"] 379 + if not isinstance(snapshot_id, str) or not snapshot_id: 380 + raise ValueError("snapshot_id must be a non-empty string") 381 + raw_files = record["files"] 382 + if not isinstance(raw_files, list) or not raw_files: 383 + raise ValueError("files must be a non-empty list") 384 + files = tuple(_parse_file_record(file) for file in raw_files) 385 + return _LedgerEvent( 386 + event_kind=EVENT_OFFLOAD, 387 + time=event_time, 388 + day=record["day"], 389 + stream=record["stream"], 390 + segment=record["segment"], 391 + snapshot_id=snapshot_id, 392 + files=files, 393 + ) 394 + 395 + if event_kind == EVENT_RESTORE: 396 + expected = {"event_kind", "time", "day", "stream", "segment"} 397 + if set(record) != expected: 398 + raise ValueError("restore record has unexpected fields") 399 + _validate_identity(record["day"], record["stream"], record["segment"]) 400 + return _LedgerEvent( 401 + event_kind=EVENT_RESTORE, 402 + time=_read_event_time(record["time"]), 403 + day=record["day"], 404 + stream=record["stream"], 405 + segment=record["segment"], 406 + snapshot_id=None, 407 + files=(), 408 + ) 409 + 410 + raise ValueError("event_kind must be offload or restore") 411 + 412 + 413 + def _parse_file_record(record: Any) -> OffloadFile: 414 + if not isinstance(record, dict) or set(record) != {"name", "bytes", "sha256"}: 415 + raise ValueError("file record must contain name, bytes, sha256") 416 + _validate_file_fields(record["name"], record["bytes"], record["sha256"]) 417 + return OffloadFile( 418 + name=record["name"], 419 + bytes=record["bytes"], 420 + sha256=record["sha256"], 421 + ) 422 + 423 + 424 + def _fold_events( 425 + events: tuple[_LedgerEvent, ...], 426 + ) -> dict[tuple[str, str, str], _LedgerEvent]: 427 + states: dict[tuple[str, str, str], _LedgerEvent] = {} 428 + for event in events: 429 + key = (event.day, event.stream, event.segment) 430 + if event.event_kind == EVENT_OFFLOAD: 431 + states[key] = event 432 + elif event.event_kind == EVENT_RESTORE: 433 + states.pop(key, None) 434 + else: # pragma: no cover - parser owns the closed vocabulary. 435 + raise RuntimeError(f"unexpected offload event kind: {event.event_kind}") 436 + return states 437 + 438 + 439 + def _segment_summary( 440 + day: str, 441 + stream: str, 442 + segment: str, 443 + current: _LedgerEvent | None, 444 + *, 445 + skipped_records: int, 446 + unreadable_ledgers: tuple[str, ...], 447 + ) -> SegmentOffloadSummary: 448 + files = current.files if current is not None else () 449 + return SegmentOffloadSummary( 450 + day=day, 451 + stream=stream, 452 + segment=segment, 453 + currently_offloaded=current is not None, 454 + snapshot_id=current.snapshot_id if current is not None else None, 455 + files=files, 456 + offloaded_bytes=sum(file.bytes for file in files), 457 + offloaded_file_count=len(files), 458 + skipped_records=skipped_records, 459 + unreadable_ledgers=unreadable_ledgers, 460 + ) 461 + 462 + 463 + def _day_summary( 464 + day: str, 465 + segments: tuple[SegmentOffloadSummary, ...], 466 + *, 467 + skipped_records: int, 468 + unreadable_ledgers: tuple[str, ...], 469 + ) -> DayOffloadSummary: 470 + return DayOffloadSummary( 471 + day=day, 472 + segments=segments, 473 + offloaded_bytes=sum(segment.offloaded_bytes for segment in segments), 474 + offloaded_file_count=sum(segment.offloaded_file_count for segment in segments), 475 + offloaded_segments=sum( 476 + 1 for segment in segments if segment.currently_offloaded 477 + ), 478 + skipped_records=skipped_records, 479 + unreadable_ledgers=unreadable_ledgers, 480 + ) 481 + 482 + 483 + __all__ = [ 484 + "DayOffloadSummary", 485 + "EVENT_OFFLOAD", 486 + "EVENT_RESTORE", 487 + "JournalOffloadSummary", 488 + "OffloadFile", 489 + "SegmentOffloadSummary", 490 + "append_offload_event", 491 + "append_restore_event", 492 + "ledger_path_for_day", 493 + "summarize_day", 494 + "summarize_journal", 495 + "summarize_segment", 496 + ]

+85

solstone/think/offload_measurement.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Read-only measurement helpers for media offload.""" 5 + 6 + from __future__ import annotations 7 + 8 + import shutil 9 + from dataclasses import dataclass 10 + from pathlib import Path 11 + 12 + from solstone.think.retention import get_raw_media_files 13 + from solstone.think.utils import day_dirs, get_journal, iter_segments 14 + 15 + MIN_FLOOR_BYTES = 20_000_000_000 16 + 17 + 18 + @dataclass(frozen=True) 19 + class RawMediaDayUsage: 20 + day: str 21 + bytes: int 22 + files: int 23 + 24 + 25 + @dataclass(frozen=True) 26 + class RawMediaUsage: 27 + total_bytes: int 28 + total_files: int 29 + per_day: tuple[RawMediaDayUsage, ...] 30 + 31 + 32 + @dataclass(frozen=True) 33 + class SuggestedOffloadDefaults: 34 + budget_bytes: int 35 + floor_bytes: int 36 + 37 + 38 + def measure_raw_media_usage() -> RawMediaUsage: 39 + per_day: list[RawMediaDayUsage] = [] 40 + total_bytes = 0 41 + total_files = 0 42 + 43 + for day in sorted(day_dirs().keys()): 44 + day_bytes = 0 45 + day_files = 0 46 + for _stream, _segment, segment_path in iter_segments(day): 47 + for raw_file in get_raw_media_files(segment_path): 48 + try: 49 + size = raw_file.stat().st_size 50 + except FileNotFoundError: 51 + continue 52 + day_bytes += size 53 + day_files += 1 54 + per_day.append(RawMediaDayUsage(day=day, bytes=day_bytes, files=day_files)) 55 + total_bytes += day_bytes 56 + total_files += day_files 57 + 58 + return RawMediaUsage( 59 + total_bytes=total_bytes, 60 + total_files=total_files, 61 + per_day=tuple(per_day), 62 + ) 63 + 64 + 65 + def device_free_bytes() -> int: 66 + return shutil.disk_usage(Path(get_journal())).free 67 + 68 + 69 + def suggest_offload_defaults(total_bytes: int) -> SuggestedOffloadDefaults: 70 + if type(total_bytes) is not int or total_bytes <= 0: 71 + raise ValueError("total_bytes must be a positive integer") 72 + budget = total_bytes // 2 73 + floor = min(max(total_bytes // 10, MIN_FLOOR_BYTES), total_bytes // 4) 74 + return SuggestedOffloadDefaults(budget_bytes=budget, floor_bytes=floor) 75 + 76 + 77 + __all__ = [ 78 + "MIN_FLOOR_BYTES", 79 + "RawMediaDayUsage", 80 + "RawMediaUsage", 81 + "SuggestedOffloadDefaults", 82 + "device_free_bytes", 83 + "measure_raw_media_usage", 84 + "suggest_offload_defaults", 85 + ]

+74

tests/test_backup_offload_guards.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from fnmatch import fnmatchcase 7 + 8 + import pytest 9 + 10 + from solstone.think.backup.engine import BACKUP_EXCLUDES 11 + 12 + 13 + def _restic_excludes(pattern: str, rel_path: str) -> bool: 14 + path_parts = tuple(part for part in rel_path.strip("/").split("/") if part) 15 + if "/" not in pattern: 16 + return any(fnmatchcase(part, pattern) for part in path_parts) 17 + 18 + anchored = pattern.startswith("/") 19 + pattern_parts = tuple(part for part in pattern.strip("/").split("/") if part) 20 + starts = (0,) if anchored else range(len(path_parts)) 21 + for start in starts: 22 + for end in range(start, len(path_parts) + 1): 23 + if _match_components(pattern_parts, path_parts[start:end]): 24 + return True 25 + return False 26 + 27 + 28 + def _match_components( 29 + pattern_parts: tuple[str, ...], path_parts: tuple[str, ...] 30 + ) -> bool: 31 + if not pattern_parts: 32 + return not path_parts 33 + head, *tail = pattern_parts 34 + rest = tuple(tail) 35 + if head == "**": 36 + return _match_components(rest, path_parts) or ( 37 + bool(path_parts) and _match_components(pattern_parts, path_parts[1:]) 38 + ) 39 + return ( 40 + bool(path_parts) 41 + and fnmatchcase(path_parts[0], head) 42 + and _match_components(rest, path_parts[1:]) 43 + ) 44 + 45 + 46 + @pytest.mark.parametrize( 47 + ("pattern", "rel_path", "expected"), 48 + [ 49 + ("health", "health/offload/20260101.jsonl", True), 50 + ("*.jsonl", "health/offload/20260101.jsonl", True), 51 + ("health/*.jsonl", "health/offload/20260101.jsonl", False), 52 + ("offload/*.jsonl", "health/offload/20260101.jsonl", True), 53 + ("/health/offload", "health/offload/20260101.jsonl", True), 54 + ("/offload/*.jsonl", "health/offload/20260101.jsonl", False), 55 + ("health/**/20260101.jsonl", "health/offload/20260101.jsonl", True), 56 + ], 57 + ) 58 + def test_restic_exclude_component_model( 59 + pattern: str, rel_path: str, expected: bool 60 + ) -> None: 61 + assert _restic_excludes(pattern, rel_path) is expected 62 + 63 + 64 + def test_offload_ledger_survives_backup_excludes() -> None: 65 + # engine.py:58-62 documents why this exists: a bare "health" pattern 66 + # previously matched health/ content by basename at every depth. 67 + ledger_rel_path = "health/offload/20260101.jsonl" 68 + 69 + assert [ 70 + pattern 71 + for pattern in BACKUP_EXCLUDES 72 + if _restic_excludes(pattern, ledger_rel_path) 73 + ] == [] 74 + assert _restic_excludes("health", ledger_rel_path)

+187

tests/test_backup_offload_state.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + import json 7 + from pathlib import Path 8 + 9 + import pytest 10 + 11 + from solstone.think.backup import state 12 + 13 + 14 + def _config_path(journal: Path) -> Path: 15 + return journal / "config" / "journal.json" 16 + 17 + 18 + def _write_config(journal: Path, payload: dict) -> None: 19 + config_path = _config_path(journal) 20 + config_path.parent.mkdir(parents=True, exist_ok=True) 21 + config_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") 22 + 23 + 24 + def _read_config(journal: Path) -> dict: 25 + return json.loads(_config_path(journal).read_text(encoding="utf-8")) 26 + 27 + 28 + def test_missing_offload_key_reads_pinned_defaults( 29 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 30 + ) -> None: 31 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 32 + _write_config(tmp_path, {"backup": {"enabled": True}}) 33 + 34 + assert state.get_backup_config()["offload"] == { 35 + "enabled": False, 36 + "budget_bytes": None, 37 + "floor_bytes": None, 38 + } 39 + 40 + 41 + def test_set_offload_valid_write_round_trips( 42 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 43 + ) -> None: 44 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 45 + _write_config(tmp_path, {}) 46 + offload = { 47 + "enabled": True, 48 + "budget_bytes": 500_000_000_000, 49 + "floor_bytes": None, 50 + } 51 + 52 + state.set_offload(offload) 53 + 54 + assert _read_config(tmp_path)["backup"]["offload"] == offload 55 + assert state.get_backup_config()["offload"] == offload 56 + 57 + 58 + @pytest.mark.parametrize( 59 + "offload", 60 + [ 61 + {"enabled": 1, "budget_bytes": None, "floor_bytes": None}, 62 + {"enabled": "true", "budget_bytes": None, "floor_bytes": None}, 63 + {"enabled": True, "budget_bytes": True, "floor_bytes": None}, 64 + {"enabled": True, "budget_bytes": False, "floor_bytes": None}, 65 + {"enabled": True, "budget_bytes": 0, "floor_bytes": None}, 66 + {"enabled": True, "budget_bytes": -1, "floor_bytes": None}, 67 + {"enabled": True, "budget_bytes": "1", "floor_bytes": None}, 68 + {"enabled": True, "budget_bytes": None, "floor_bytes": 0}, 69 + {"enabled": True, "budget_bytes": None, "floor_bytes": -1}, 70 + {"enabled": True, "budget_bytes": None, "floor_bytes": "1"}, 71 + {"enabled": True, "budget_bytes": None}, 72 + { 73 + "enabled": True, 74 + "budget_bytes": None, 75 + "floor_bytes": None, 76 + "extra": 1, 77 + }, 78 + ], 79 + ) 80 + def test_set_offload_rejects_invalid_shapes_without_writing( 81 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, offload: dict 82 + ) -> None: 83 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 84 + original = {"backup": {"enabled": False}} 85 + _write_config(tmp_path, original) 86 + before = _read_config(tmp_path) 87 + 88 + with pytest.raises(ValueError): 89 + state.set_offload(offload) 90 + 91 + assert _read_config(tmp_path) == before 92 + 93 + 94 + def test_set_offload_preserves_existing_backup_section_without_materializing_defaults( 95 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 96 + ) -> None: 97 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 98 + backup = { 99 + "enabled": False, 100 + "mode": "operated", 101 + "retention": {"hourly": 2, "daily": 3, "weekly": 4, "monthly": 5}, 102 + "custom_marker": {"nested": ["keep", 7]}, 103 + } 104 + _write_config(tmp_path, {"backup": backup}) 105 + before_backup = _read_config(tmp_path)["backup"] 106 + 107 + state.set_offload( 108 + { 109 + "enabled": True, 110 + "budget_bytes": 10, 111 + "floor_bytes": 5, 112 + } 113 + ) 114 + 115 + after_backup = _read_config(tmp_path)["backup"] 116 + assert set(after_backup) == {*before_backup, "offload"} 117 + for key, value in before_backup.items(): 118 + assert json.dumps(after_backup[key], sort_keys=True) == json.dumps( 119 + value, sort_keys=True 120 + ) 121 + assert "destination" not in after_backup 122 + assert "last_prune" not in after_backup 123 + 124 + 125 + def test_set_offload_has_no_cross_field_backup_enabled_rule( 126 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 127 + ) -> None: 128 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 129 + _write_config(tmp_path, {"backup": {"enabled": False}}) 130 + 131 + state.set_offload( 132 + { 133 + "enabled": True, 134 + "budget_bytes": 10, 135 + "floor_bytes": 20, 136 + } 137 + ) 138 + 139 + assert _read_config(tmp_path)["backup"]["enabled"] is False 140 + assert _read_config(tmp_path)["backup"]["offload"] == { 141 + "enabled": True, 142 + "budget_bytes": 10, 143 + "floor_bytes": 20, 144 + } 145 + 146 + 147 + def test_offload_state_records_and_status_view( 148 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 149 + ) -> None: 150 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 151 + _write_config(tmp_path, {"backup": {}}) 152 + 153 + config = state.get_backup_config() 154 + assert config["last_offload"] == {"time": None, "status": None, "reason": None} 155 + assert config["last_verification"] == { 156 + "time": None, 157 + "status": None, 158 + "reason": None, 159 + } 160 + 161 + state.record_offload_result(status="stalled", time=123, reason="no_progress") 162 + state.record_verification_result(status="skipped", time=None, reason="disabled") 163 + 164 + backup = _read_config(tmp_path)["backup"] 165 + assert backup["last_offload"] == { 166 + "time": 123, 167 + "status": "stalled", 168 + "reason": "no_progress", 169 + } 170 + assert backup["last_verification"] == { 171 + "time": None, 172 + "status": "skipped", 173 + "reason": "disabled", 174 + } 175 + with pytest.raises(ValueError): 176 + state.record_offload_result(status="degraded", time=1) 177 + with pytest.raises(ValueError): 178 + state.record_verification_result(status="stalled", time=1) 179 + 180 + view = state.status_view() 181 + assert view["offload"] == { 182 + "enabled": False, 183 + "budget_bytes": None, 184 + "floor_bytes": None, 185 + } 186 + assert view["last_offload"] == backup["last_offload"] 187 + assert view["last_verification"] == backup["last_verification"]

+390

tests/test_offload_ledger.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + import json 7 + import logging 8 + import os 9 + import stat 10 + from collections.abc import Callable 11 + from pathlib import Path 12 + 13 + import pytest 14 + 15 + from solstone.think.offload_ledger import ( 16 + EVENT_OFFLOAD, 17 + EVENT_RESTORE, 18 + OffloadFile, 19 + append_offload_event, 20 + append_restore_event, 21 + ledger_path_for_day, 22 + summarize_day, 23 + summarize_journal, 24 + summarize_segment, 25 + ) 26 + 27 + DAY = "20260101" 28 + STREAM = "archon" 29 + SEGMENT = "120000_300" 30 + SHA_A = "a" * 64 31 + SHA_B = "b" * 64 32 + 33 + 34 + def _is_ledger_fd_fsynced( 35 + monkeypatch: pytest.MonkeyPatch, ledger_path: Path, writer: Callable[[], None] 36 + ) -> bool: 37 + calls = [] 38 + real_fsync = os.fsync 39 + 40 + def spy(fd: int) -> None: 41 + fd_stat = os.fstat(fd) 42 + try: 43 + ledger_stat = ledger_path.stat() 44 + except FileNotFoundError: 45 + ledger_stat = None 46 + if ( 47 + ledger_stat is not None 48 + and stat.S_ISREG(fd_stat.st_mode) 49 + and (fd_stat.st_dev, fd_stat.st_ino) 50 + == (ledger_stat.st_dev, ledger_stat.st_ino) 51 + ): 52 + calls.append(fd) 53 + real_fsync(fd) 54 + 55 + monkeypatch.setattr("solstone.think.journal_io.append.os.fsync", spy) 56 + writer() 57 + return bool(calls) 58 + 59 + 60 + def _write_plain_jsonl(path: Path, record: dict) -> None: 61 + path.parent.mkdir(parents=True, exist_ok=True) 62 + with open(path, "ab") as handle: 63 + handle.write((json.dumps(record) + "\n").encode("utf-8")) 64 + handle.flush() 65 + 66 + 67 + def _offload_record( 68 + *, 69 + day: str, 70 + stream: str, 71 + segment: str, 72 + snapshot_id: str, 73 + size: int, 74 + name: str = "audio.wav", 75 + sha256: str = SHA_A, 76 + time: int = 1, 77 + ) -> dict: 78 + return { 79 + "event_kind": EVENT_OFFLOAD, 80 + "time": time, 81 + "day": day, 82 + "stream": stream, 83 + "segment": segment, 84 + "snapshot_id": snapshot_id, 85 + "files": [{"name": name, "bytes": size, "sha256": sha256}], 86 + } 87 + 88 + 89 + def _restore_record( 90 + *, 91 + day: str, 92 + stream: str, 93 + segment: str, 94 + time: int = 1, 95 + ) -> dict: 96 + return { 97 + "event_kind": EVENT_RESTORE, 98 + "time": time, 99 + "day": day, 100 + "stream": stream, 101 + "segment": segment, 102 + } 103 + 104 + 105 + def test_offload_append_writes_media_day_ledger_and_segment_summary( 106 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 107 + ) -> None: 108 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 109 + 110 + append_offload_event( 111 + day=DAY, 112 + stream=STREAM, 113 + segment=SEGMENT, 114 + snapshot_id="snap-1", 115 + files=[ 116 + OffloadFile(name="audio.wav", bytes=10, sha256=SHA_A), 117 + OffloadFile(name="screen.mp4", bytes=20, sha256=SHA_B), 118 + ], 119 + time=10, 120 + ) 121 + 122 + ledger_path = ledger_path_for_day(DAY) 123 + assert ledger_path == tmp_path / "health" / "offload" / f"{DAY}.jsonl" 124 + record = json.loads(ledger_path.read_text(encoding="utf-8").splitlines()[0]) 125 + assert "schema_version" not in record 126 + assert record["event_kind"] == EVENT_OFFLOAD 127 + assert record["snapshot_id"] == "snap-1" 128 + assert record["files"] == [ 129 + {"name": "audio.wav", "bytes": 10, "sha256": SHA_A}, 130 + {"name": "screen.mp4", "bytes": 20, "sha256": SHA_B}, 131 + ] 132 + assert "hash" not in record["files"][0] 133 + 134 + summary = summarize_segment(DAY, STREAM, SEGMENT) 135 + assert summary.currently_offloaded is True 136 + assert summary.snapshot_id == "snap-1" 137 + assert summary.offloaded_bytes == 30 138 + assert summary.offloaded_file_count == 2 139 + assert summary.files[0].sha256 == SHA_A 140 + 141 + 142 + def test_restore_and_reoffload_fold_by_append_order_not_time( 143 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 144 + ) -> None: 145 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 146 + 147 + append_offload_event( 148 + day=DAY, 149 + stream=STREAM, 150 + segment=SEGMENT, 151 + snapshot_id="snap-old", 152 + files=[OffloadFile(name="audio.wav", bytes=10, sha256=SHA_A)], 153 + time=300, 154 + ) 155 + append_restore_event(day=DAY, stream=STREAM, segment=SEGMENT, time=200) 156 + 157 + restored = summarize_segment(DAY, STREAM, SEGMENT) 158 + assert restored.currently_offloaded is False 159 + assert restored.offloaded_bytes == 0 160 + assert restored.snapshot_id is None 161 + 162 + append_offload_event( 163 + day=DAY, 164 + stream=STREAM, 165 + segment=SEGMENT, 166 + snapshot_id="snap-new", 167 + files=[OffloadFile(name="audio.wav", bytes=11, sha256=SHA_B)], 168 + time=100, 169 + ) 170 + 171 + reoffloaded = summarize_segment(DAY, STREAM, SEGMENT) 172 + assert reoffloaded.currently_offloaded is True 173 + assert reoffloaded.snapshot_id == "snap-new" 174 + assert reoffloaded.offloaded_bytes == 11 175 + 176 + 177 + def test_append_fsyncs_ledger_fd_before_return( 178 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 179 + ) -> None: 180 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 181 + ledger_path = ledger_path_for_day(DAY) 182 + 183 + with monkeypatch.context() as context: 184 + assert _is_ledger_fd_fsynced( 185 + context, 186 + ledger_path, 187 + lambda: append_offload_event( 188 + day=DAY, 189 + stream=STREAM, 190 + segment=SEGMENT, 191 + snapshot_id="snap-1", 192 + files=[OffloadFile(name="audio.wav", bytes=10, sha256=SHA_A)], 193 + time=1, 194 + ), 195 + ) 196 + 197 + plain_path = tmp_path / "health" / "offload" / "plain.jsonl" 198 + with monkeypatch.context() as context: 199 + assert not _is_ledger_fd_fsynced( 200 + context, 201 + plain_path, 202 + lambda: _write_plain_jsonl( 203 + plain_path, 204 + _offload_record( 205 + day=DAY, 206 + stream=STREAM, 207 + segment=SEGMENT, 208 + snapshot_id="plain", 209 + size=10, 210 + ), 211 + ), 212 + ) 213 + 214 + 215 + def test_day_and_journal_summaries_fold_mixed_days( 216 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 217 + ) -> None: 218 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 219 + append_offload_event( 220 + day="20260101", 221 + stream=STREAM, 222 + segment="120000_300", 223 + snapshot_id="full", 224 + files=[ 225 + OffloadFile(name="a.wav", bytes=10, sha256=SHA_A), 226 + OffloadFile(name="b.mp4", bytes=5, sha256=SHA_B), 227 + ], 228 + time=1, 229 + ) 230 + append_offload_event( 231 + day="20260102", 232 + stream=STREAM, 233 + segment="120000_300", 234 + snapshot_id="restored", 235 + files=[OffloadFile(name="a.wav", bytes=20, sha256=SHA_A)], 236 + time=1, 237 + ) 238 + append_restore_event(day="20260102", stream=STREAM, segment="120000_300", time=2) 239 + append_offload_event( 240 + day="20260103", 241 + stream=STREAM, 242 + segment="120000_300", 243 + snapshot_id="mixed-kept", 244 + files=[OffloadFile(name="a.wav", bytes=30, sha256=SHA_A)], 245 + time=1, 246 + ) 247 + append_offload_event( 248 + day="20260103", 249 + stream=STREAM, 250 + segment="121000_300", 251 + snapshot_id="mixed-restored", 252 + files=[OffloadFile(name="b.wav", bytes=40, sha256=SHA_B)], 253 + time=1, 254 + ) 255 + append_restore_event(day="20260103", stream=STREAM, segment="121000_300", time=2) 256 + 257 + assert summarize_day("20260101").offloaded_bytes == 15 258 + assert summarize_day("20260102").offloaded_bytes == 0 259 + mixed = summarize_day("20260103") 260 + assert mixed.offloaded_bytes == 30 261 + assert mixed.offloaded_segments == 1 262 + 263 + journal = summarize_journal() 264 + assert [day.day for day in journal.days] == ["20260101", "20260102", "20260103"] 265 + assert journal.offloaded_bytes == 45 266 + assert journal.offloaded_file_count == 3 267 + assert journal.offloaded_segments == 2 268 + assert journal.offloaded_days == 2 269 + 270 + 271 + def test_malformed_line_warns_counts_skipped_and_keeps_valid_totals( 272 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture 273 + ) -> None: 274 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 275 + ledger_path = ledger_path_for_day(DAY) 276 + ledger_path.parent.mkdir(parents=True) 277 + ledger_path.write_text( 278 + "\n".join( 279 + [ 280 + json.dumps( 281 + _offload_record( 282 + day=DAY, 283 + stream=STREAM, 284 + segment="120000_300", 285 + snapshot_id="snap-1", 286 + size=10, 287 + ) 288 + ), 289 + "{bad", 290 + json.dumps( 291 + _offload_record( 292 + day=DAY, 293 + stream=STREAM, 294 + segment="121000_300", 295 + snapshot_id="snap-2", 296 + size=20, 297 + sha256=SHA_B, 298 + ) 299 + ), 300 + ] 301 + ) 302 + + "\n", 303 + encoding="utf-8", 304 + ) 305 + 306 + with caplog.at_level(logging.WARNING, logger="solstone.think.offload_ledger"): 307 + summary = summarize_day(DAY) 308 + 309 + assert summary.offloaded_bytes == 30 310 + assert summary.skipped_records == 1 311 + assert any(str(ledger_path) in record.message for record in caplog.records) 312 + assert any("malformed" in record.message for record in caplog.records) 313 + 314 + 315 + def test_null_time_record_is_skipped_not_fabricated( 316 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture 317 + ) -> None: 318 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 319 + ledger_path = ledger_path_for_day(DAY) 320 + invalid = _offload_record( 321 + day=DAY, 322 + stream=STREAM, 323 + segment="120000_300", 324 + snapshot_id="bad-time", 325 + size=20, 326 + ) 327 + invalid["time"] = None 328 + ledger_path.parent.mkdir(parents=True) 329 + ledger_path.write_text( 330 + "\n".join( 331 + [ 332 + json.dumps( 333 + _offload_record( 334 + day=DAY, 335 + stream=STREAM, 336 + segment="121000_300", 337 + snapshot_id="valid", 338 + size=10, 339 + ) 340 + ), 341 + json.dumps(invalid), 342 + ] 343 + ) 344 + + "\n", 345 + encoding="utf-8", 346 + ) 347 + 348 + with caplog.at_level(logging.WARNING, logger="solstone.think.offload_ledger"): 349 + summary = summarize_day(DAY) 350 + 351 + assert summary.offloaded_bytes == 10 352 + assert summary.skipped_records == 1 353 + assert any(str(ledger_path) in record.message for record in caplog.records) 354 + 355 + 356 + def test_undecodable_ledger_degrades_without_clean_zero( 357 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture 358 + ) -> None: 359 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 360 + ledger_path = ledger_path_for_day(DAY) 361 + ledger_path.parent.mkdir(parents=True) 362 + ledger_path.write_bytes(b"\xff") 363 + 364 + with caplog.at_level(logging.WARNING, logger="solstone.think.offload_ledger"): 365 + summary = summarize_day(DAY) 366 + 367 + assert summary.offloaded_bytes == 0 368 + assert summary.degraded is True 369 + assert summary.unreadable_ledgers == (str(ledger_path),) 370 + assert any(str(ledger_path) in record.message for record in caplog.records) 371 + 372 + 373 + def test_absent_or_empty_ledger_is_clean_zero( 374 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 375 + ) -> None: 376 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 377 + 378 + missing = summarize_journal() 379 + assert missing.offloaded_bytes == 0 380 + assert missing.skipped_records == 0 381 + assert missing.degraded is False 382 + 383 + ledger_path = ledger_path_for_day(DAY) 384 + ledger_path.parent.mkdir(parents=True) 385 + ledger_path.write_text("", encoding="utf-8") 386 + 387 + empty = summarize_day(DAY) 388 + assert empty.offloaded_bytes == 0 389 + assert empty.skipped_records == 0 390 + assert empty.degraded is False

+123

tests/test_offload_measurement.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from __future__ import annotations 5 + 6 + from pathlib import Path 7 + from types import SimpleNamespace 8 + 9 + import pytest 10 + 11 + from solstone.think import offload_measurement 12 + from solstone.think.offload_measurement import ( 13 + device_free_bytes, 14 + measure_raw_media_usage, 15 + suggest_offload_defaults, 16 + ) 17 + from solstone.think.retention import compute_storage_summary 18 + 19 + GB = 10**9 20 + 21 + 22 + def _segment(journal: Path, day: str, stream: str = "archon") -> Path: 23 + path = journal / "chronicle" / day / stream / "120000_300" 24 + path.mkdir(parents=True, exist_ok=True) 25 + return path 26 + 27 + 28 + def _write_bytes(path: Path, size: int) -> Path: 29 + path.parent.mkdir(parents=True, exist_ok=True) 30 + path.write_bytes(b"x" * size) 31 + return path 32 + 33 + 34 + def test_raw_media_measurement_matches_retention_predicate_non_recursive( 35 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 36 + ) -> None: 37 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 38 + segment = _segment(tmp_path, "20260101") 39 + _write_bytes(segment / "audio.wav", 10) 40 + _write_bytes(segment / "clip.mp4", 20) 41 + _write_bytes(segment / "frame.png", 30) 42 + _write_bytes(segment / "monitor_0_diff.png", 40) 43 + _write_bytes(segment / "audio.jsonl", 50) 44 + _write_bytes(segment / "note.json", 60) 45 + _write_bytes(segment / "talents" / "frame.png", 70) 46 + 47 + usage = measure_raw_media_usage() 48 + 49 + assert usage.total_bytes == 100 50 + assert usage.total_files == 4 51 + assert usage.per_day == (offload_measurement.RawMediaDayUsage("20260101", 100, 4),) 52 + assert compute_storage_summary().raw_media_bytes == usage.total_bytes 53 + 54 + 55 + def test_raw_media_per_day_breakdown_is_chronological( 56 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 57 + ) -> None: 58 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 59 + _write_bytes(_segment(tmp_path, "20260301") / "audio.wav", 3) 60 + _write_bytes(_segment(tmp_path, "20260101") / "audio.wav", 1) 61 + _write_bytes(_segment(tmp_path, "20260201") / "audio.wav", 2) 62 + 63 + usage = measure_raw_media_usage() 64 + 65 + assert [day.day for day in usage.per_day] == [ 66 + "20260101", 67 + "20260201", 68 + "20260301", 69 + ] 70 + assert [day.bytes for day in usage.per_day] == [1, 2, 3] 71 + 72 + 73 + def test_raw_media_measurement_tolerates_file_vanishing_before_stat( 74 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 75 + ) -> None: 76 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 77 + segment = _segment(tmp_path, "20260101") 78 + vanishing = _write_bytes(segment / "vanishing.wav", 10) 79 + stable = _write_bytes(segment / "stable.wav", 20) 80 + 81 + def fake_get_raw_media_files(_segment_path: Path) -> list[Path]: 82 + vanishing.unlink() 83 + return [vanishing, stable] 84 + 85 + monkeypatch.setattr( 86 + offload_measurement, "get_raw_media_files", fake_get_raw_media_files 87 + ) 88 + 89 + usage = measure_raw_media_usage() 90 + 91 + assert usage.total_bytes == 20 92 + assert usage.total_files == 1 93 + 94 + 95 + def test_device_free_bytes_uses_disk_usage_free( 96 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch 97 + ) -> None: 98 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 99 + 100 + def fake_disk_usage(path: Path) -> SimpleNamespace: 101 + assert path == tmp_path 102 + return SimpleNamespace(total=1000 * GB, used=100 * GB, free=850 * GB) 103 + 104 + monkeypatch.setattr(offload_measurement.shutil, "disk_usage", fake_disk_usage) 105 + 106 + assert device_free_bytes() == 850 * GB 107 + 108 + 109 + @pytest.mark.parametrize( 110 + ("total", "budget", "floor"), 111 + [ 112 + (1000 * GB, 500 * GB, 100 * GB), 113 + (100 * GB, 50 * GB, 20 * GB), 114 + (30 * GB, 15 * GB, 7_500_000_000), 115 + ], 116 + ) 117 + def test_suggest_offload_defaults_decimal_gb( 118 + total: int, budget: int, floor: int 119 + ) -> None: 120 + defaults = suggest_offload_defaults(total) 121 + 122 + assert defaults.budget_bytes == budget 123 + assert defaults.floor_bytes == floor

Configure Feed

Configure Feed