Spaces:
Paused
Paused
File size: 6,399 Bytes
5e0b58b | 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 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | """Crash-safe generation compaction for Brain-5D persistence."""
from __future__ import annotations
import json
import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from .b5d import B5DReader
from .delta_journal import DeltaJournal
from .recovery import RecoveryManager
@dataclass(frozen=True, slots=True)
class StorageGeneration:
"""One snapshot/journal generation selected by an atomic manifest."""
generation: int
snapshot: str
journal: str
base_tick: int
@dataclass(frozen=True, slots=True)
class CompactionResult:
"""Result of one successful or skipped compaction."""
compacted: bool
generation: int
base_tick: int
snapshot_path: Path
journal_path: Path
manifest_path: Path
class StorageManifest:
"""Atomic pointer to the active snapshot/journal generation."""
def __init__(self, path: Path) -> None:
self.path = path
def read(self) -> StorageGeneration:
"""Read the currently active storage generation."""
raw_data = json.loads(self.path.read_text(encoding="utf-8"))
if not isinstance(raw_data, dict):
raise ValueError("invalid storage manifest: root must be a dict")
raw = cast(dict[str, Any], raw_data)
if raw.get("version") != 1:
raise ValueError("invalid storage manifest: version must be 1")
# Extract values with explicit type conversion
generation_raw = raw.get("generation")
if not isinstance(generation_raw, int):
raise ValueError("manifest missing 'generation' field")
generation = int(generation_raw)
snapshot = raw.get("snapshot")
if not isinstance(snapshot, str):
raise ValueError("manifest missing 'snapshot' field")
snapshot = str(snapshot)
journal = raw.get("journal")
if not isinstance(journal, str):
raise ValueError("manifest missing 'journal' field")
journal = str(journal)
base_tick_raw = raw.get("base_tick")
if not isinstance(base_tick_raw, int):
raise ValueError("manifest missing 'base_tick' field")
base_tick = int(base_tick_raw)
return StorageGeneration(
generation=generation,
snapshot=snapshot,
journal=journal,
base_tick=base_tick,
)
def write_atomic(self, generation: StorageGeneration) -> None:
"""Atomically publish one fully prepared storage generation."""
self.path.parent.mkdir(parents=True, exist_ok=True)
payload = json.dumps(
{
"version": 1,
"generation": generation.generation,
"snapshot": generation.snapshot,
"journal": generation.journal,
"base_tick": generation.base_tick,
},
sort_keys=True,
separators=(",", ":"),
)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
prefix=self.path.name + ".",
suffix=".tmp",
dir=self.path.parent,
delete=False,
) as handle:
temporary = Path(handle.name)
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
try:
os.replace(temporary, self.path)
finally:
if temporary.exists():
temporary.unlink()
class StorageCompactor:
"""Compact committed journal state into a new immutable generation."""
def __init__(self, root: Path, stem: str = "brain5d") -> None:
self.root = root
self.stem = stem
self.manifest = StorageManifest(root / f"{stem}.manifest.json")
def initialize(self, snapshot_path: Path, journal_path: Path) -> StorageGeneration:
"""Create generation zero manifest for an existing pair."""
with B5DReader(snapshot_path) as reader:
base_tick = reader.header.snapshot_tick
generation = StorageGeneration(
generation=0,
snapshot=snapshot_path.name,
journal=journal_path.name,
base_tick=base_tick,
)
self.manifest.write_atomic(generation)
return generation
def compact(self) -> CompactionResult:
"""Recover committed deltas into a new generation and publish atomically."""
active = self.manifest.read()
snapshot_path = self.root / active.snapshot
journal_path = self.root / active.journal
with DeltaJournal(journal_path, base_tick=active.base_tick) as journal:
scan = journal.validate()
marker = scan.last_commit
if marker is None:
return CompactionResult(
compacted=False,
generation=active.generation,
base_tick=active.base_tick,
snapshot_path=snapshot_path,
journal_path=journal_path,
manifest_path=self.manifest.path,
)
committed_tick = marker.tick
next_generation = active.generation + 1
next_snapshot = self.root / f"{self.stem}.g{next_generation}.b5d"
next_journal = self.root / f"{self.stem}.g{next_generation}.b5d.journal"
recovery = RecoveryManager(snapshot_path, journal_path).recover(next_snapshot)
if not recovery.success:
raise RuntimeError(recovery.error or "compaction recovery failed")
with B5DReader(next_snapshot) as reader:
reader.validate_invariants()
if reader.header.snapshot_tick != committed_tick:
raise RuntimeError("compacted snapshot tick does not match commit")
with DeltaJournal(next_journal, base_tick=committed_tick):
pass
published = StorageGeneration(
generation=next_generation,
snapshot=next_snapshot.name,
journal=next_journal.name,
base_tick=committed_tick,
)
self.manifest.write_atomic(published)
return CompactionResult(
compacted=True,
generation=next_generation,
base_tick=committed_tick,
snapshot_path=next_snapshot,
journal_path=next_journal,
manifest_path=self.manifest.path,
)
|