diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index a0c7abbce9..0a9ea5c36e 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -14,6 +14,7 @@ import uuid from contextlib import closing from dataclasses import asdict, dataclass +from enum import StrEnum from pathlib import Path from types import TracebackType @@ -44,7 +45,50 @@ #: (~35 GB on the reference archive), so keeping more is expensive storage, #: not cheap insurance. SUPERSEDED_GENERATION_RETENTION = 1 +# Keep the current promotion receipt plus the immediately preceding one. This +# is intentionally larger than the rollback-generation boundary so automatic +# receipt pruning cannot erase the evidence for the active boundary. +RETENTION_RECEIPT_HISTORY = SUPERSEDED_GENERATION_RETENTION + 1 _GENERATIONS_DIRNAME = ".index-generations" +_RETENTION_RECEIPTS_DIRNAME = "retention-receipts" + + +class GenerationRetentionState(StrEnum): + """Durable lifecycle states for a promoted generation's retention record.""" + + ACTIVE = "active" + RETAINED = "retained" + ELIGIBLE = "eligible" + RECLAIMED = "reclaimed" + + +@dataclass(frozen=True, slots=True) +class GenerationRetentionRecord: + generation_id: str + generation_owner_id: str + retention_owner_id: str + state: GenerationRetentionState + + +@dataclass(frozen=True, slots=True) +class GenerationRetentionReceipt: + """Evidence for automatic rollback retention and generation reclamation.""" + + promoted_generation_id: str + promoted_at_ns: int + retention_boundary: int + automatic: bool + records: tuple[GenerationRetentionRecord, ...] + eligible_generation_ids: tuple[str, ...] = () + reclaimed_marker_count: int = 0 + + @property + def states_by_generation_id(self) -> dict[str, str]: + return {record.generation_id: record.state.value for record in self.records} + + @property + def owner_by_generation_id(self) -> dict[str, str]: + return {record.generation_id: record.retention_owner_id for record in self.records} def _is_generation_member(path: Path) -> bool: @@ -91,6 +135,14 @@ class IndexGeneration: state: str created_at_ms: int source_snapshot: str = "" + # Millisecond creation time remains for compatibility with existing + # generation metadata. Lifecycle ordering uses these nanosecond values so + # UUID text never decides which rollback target is newest. + created_at_ns: int = 0 + promoted_at_ns: int = 0 + predecessor_generation_id: str | None = None + retention_owner_id: str | None = None + retention_state: str | None = None @dataclass(frozen=True, slots=True) @@ -678,7 +730,8 @@ def next_raw_page( return RebuildRawPage(rows=tuple(selected), has_more=has_more, deferred_reason=deferred_reason) def create(self, *, owner_id: str | None = None, source_snapshot: str) -> IndexGeneration: - generation_id = f"gen-{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}" + created_at_ns = self._next_lifecycle_timestamp_ns() + generation_id = f"gen-{created_at_ns // 1_000_000}-{uuid.uuid4().hex[:8]}" owner = owner_id or str(uuid.uuid4()) root = self.generations_root / generation_id root.mkdir(parents=True, exist_ok=False) @@ -694,8 +747,9 @@ def create(self, *, owner_id: str | None = None, source_snapshot: str) -> IndexG archive_root=str(self.archive_root.resolve(strict=False)), index_path=str(index_path), state="inactive", - created_at_ms=int(time.time() * 1000), + created_at_ms=created_at_ns // 1_000_000, source_snapshot=source_snapshot, + created_at_ns=created_at_ns, ) self._write(generation) return generation @@ -708,9 +762,11 @@ def promote(self, generation: IndexGeneration) -> IndexGeneration: current = self.load(generation.generation_id) if current.owner_id != generation.owner_id or current.state != "inactive": raise RuntimeError("only the owning inactive generation can be promoted") + self._validate_retention_ownership() target = Path(current.index_path).resolve(strict=True) _checkpoint_truncate(target, label="new index") pointer = self.active_pointer + predecessor_generation_id = self._generation_id_for_active_target(pointer) retired = self.generations_root / f"retired-{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}" retired.mkdir(parents=True, exist_ok=False) if pointer.exists() or pointer.is_symlink(): @@ -724,81 +780,159 @@ def promote(self, generation: IndexGeneration) -> IndexGeneration: if pointer.exists() or pointer.is_symlink(): os.link(pointer, retired / "index.db", follow_symlinks=False) _fsync_directory(retired) - promoting = IndexGeneration(**{**asdict(current), "state": "promoting"}) + promoting = IndexGeneration( + **{ + **asdict(current), + "state": "promoting", + "predecessor_generation_id": predecessor_generation_id, + } + ) self._write(promoting) temporary = pointer.parent / f".index.db.promote-{uuid.uuid4().hex}" temporary.symlink_to(target) os.replace(temporary, pointer) _fsync_directory(pointer.parent) - promoted = IndexGeneration(**{**asdict(current), "state": "active"}) + promoted = IndexGeneration( + **{ + **asdict(current), + "state": "active", + "promoted_at_ns": self._next_lifecycle_timestamp_ns(), + "predecessor_generation_id": predecessor_generation_id, + "retention_owner_id": current.generation_id, + "retention_state": GenerationRetentionState.ACTIVE.value, + } + ) self._write(promoted) - # Housekeeping only: a failure here must not undo a promotion that has - # already swapped the pointer and written active metadata. + # Retention collection is part of promotion rather than a separate + # cleanup surface. Its receipt is written before any eligible + # generation is removed, so a completed pointer swap never reclaims + # history without durable evidence of the retention boundary. try: - self.prune_superseded_generations() + self._collect_superseded_generations(promoted) except OSError: - logger.warning("index generation pruning failed after promotion", exc_info=True) + logger.warning("index generation retention collection failed after promotion", exc_info=True) return promoted - def prune_superseded_generations(self, *, keep: int = SUPERSEDED_GENERATION_RETENTION) -> list[str]: - """Delete superseded generations beyond the retention window. + def _validate_retention_ownership(self) -> None: + """Require every prior promoted generation to name its build owner. - A promoted generation is ~35 GB. Before this existed nothing ever - removed one: ``promote`` retires the *pointer* into a ``retired-*`` - marker (a hardlink of the symlink, a few KB) but left the superseded - ``gen-*`` directory in place forever, and ``discard_if_inactive`` only - disposes of candidates that were never promoted. A live archive had - accumulated nine dead generations, ~290 GB (polylogue-wmft). + This runs before the active pointer moves. An ownerless predecessor is + ambiguous history, not a reclaimable candidate: promotion stops while + the old generation is still live rather than creating a future GC path + with no accountable owner. + """ + for metadata_path in sorted(self.generations_root.glob("gen-*/generation.json")): + try: + generation = IndexGeneration(**json.loads(metadata_path.read_text(encoding="utf-8"))) + except (OSError, ValueError, TypeError): + continue + if generation.state == "active" and not generation.owner_id.strip(): + raise RuntimeError(f"retention ownership is missing for generation {generation.generation_id}") - Retention is expressed in generations rather than bytes or age because - the reason to keep one is rollback: ``keep=1`` leaves exactly the - previous index reachable if a promotion turns out to be bad. + def _collect_superseded_generations(self, promoted: IndexGeneration) -> GenerationRetentionReceipt: + """Automatically retain one rollback target and reclaim older history. - Fails closed in every ambiguous case -- anything that is or might be - the active target, anything mid-promotion, and anything whose metadata - cannot be read is retained, never deleted. + A promoted generation is large enough that a bounded retention window + matters, but the immediately preceding generation remains rollback + capable until the next promotion crosses the declared boundary. The + receipt first records every eligible generation, then records its + reclaimed state after filesystem removal. """ - if keep < 0: - raise ValueError("keep must be non-negative") - try: - active_target = self.active_pointer.resolve(strict=True) - except OSError: - # No resolvable active index: refuse to delete anything, since the - # thing that would tell us what is live is exactly what is missing. - return [] - - candidates: list[tuple[int, str, Path]] = [] + active_target = self.active_pointer.resolve(strict=True) + candidates: list[tuple[int, int, str, Path, IndexGeneration]] = [] for metadata_path in sorted(self.generations_root.glob("gen-*/generation.json")): try: generation = IndexGeneration(**json.loads(metadata_path.read_text(encoding="utf-8"))) except (OSError, ValueError, TypeError): - continue # unreadable metadata: retain - # ONLY previously-promoted generations are superseded history. - # `state == "inactive"` means never promoted -- which is exactly - # what an in-flight or paused resumable rebuild candidate looks - # like (see `create_transaction`). Treating those as prunable let - # an unrelated promotion delete a rebuild in progress, and let a - # newer inactive candidate consume the single retained slot so the - # real rollback target went instead. Never-promoted candidates - # belong to `discard_if_inactive`, which their owner drives. + continue # unreadable metadata remains retained, never reclaimed if generation.state != "active": continue try: if Path(generation.index_path).resolve(strict=True) == active_target: continue except OSError: - # index.db already gone; the directory is still reclaimable. - pass - candidates.append((generation.created_at_ms, generation.generation_id, metadata_path.parent)) - - # generation_id breaks ties: two generations can share a millisecond, - # and a stable sort would otherwise fall back to glob order, making - # "newest" non-deterministic and the retained slot arbitrary. - candidates.sort(key=lambda item: (item[0], item[1]), reverse=True) - removed: list[str] = [] - for _created_at_ms, generation_id, directory in candidates[keep:]: + continue # an incomplete candidate remains retained + candidates.append( + ( + _generation_lifecycle_recency_ns(generation), + _generation_creation_recency_ns(generation), + generation.generation_id, + metadata_path.parent, + generation, + ) + ) + + # Promotion order decides rollback capability. A normal promotion + # records its actual predecessor before the pointer swap and pins it + # first; recovered promotions have no pre-swap observation, so their + # persisted promotion timestamp supplies the same chronology. UUIDs + # are only a deterministic final tie-break, never the recency signal. + candidates.sort(key=lambda item: (item[0], item[1], item[2]), reverse=True) + predecessor = promoted.predecessor_generation_id + if predecessor is not None: + candidates.sort(key=lambda item: item[4].generation_id != predecessor) + retained = candidates[:SUPERSEDED_GENERATION_RETENTION] + eligible = candidates[SUPERSEDED_GENERATION_RETENTION:] + records = [ + GenerationRetentionRecord( + generation_id=promoted.generation_id, + generation_owner_id=promoted.owner_id, + retention_owner_id=promoted.generation_id, + state=GenerationRetentionState.ACTIVE, + ) + ] + for _lifecycle_at_ns, _created_at_ns, _generation_id, _directory, generation in retained: + retained_generation = IndexGeneration( + **{ + **asdict(generation), + "retention_owner_id": promoted.generation_id, + "retention_state": GenerationRetentionState.RETAINED.value, + } + ) + self._write(retained_generation) + records.append( + GenerationRetentionRecord( + generation_id=retained_generation.generation_id, + generation_owner_id=retained_generation.owner_id, + retention_owner_id=promoted.generation_id, + state=GenerationRetentionState.RETAINED, + ) + ) + for _lifecycle_at_ns, _created_at_ns, _generation_id, _directory, generation in eligible: + eligible_generation = IndexGeneration( + **{ + **asdict(generation), + "retention_owner_id": promoted.generation_id, + "retention_state": GenerationRetentionState.ELIGIBLE.value, + } + ) + self._write(eligible_generation) + receipt = GenerationRetentionReceipt( + promoted_generation_id=promoted.generation_id, + promoted_at_ns=promoted.promoted_at_ns, + retention_boundary=SUPERSEDED_GENERATION_RETENTION, + automatic=True, + records=tuple(records) + + tuple( + GenerationRetentionRecord( + generation_id=generation.generation_id, + generation_owner_id=generation.owner_id, + retention_owner_id=promoted.generation_id, + state=GenerationRetentionState.ELIGIBLE, + ) + for _lifecycle_at_ns, _created_at_ns, _generation_id, _directory, generation in eligible + ), + eligible_generation_ids=tuple( + generation.generation_id + for _lifecycle_at_ns, _created_at_ns, _generation_id, _directory, generation in eligible + ), + ) + self._write_retention_receipt(receipt) + + reclaimed: list[str] = [] + for _lifecycle_at_ns, _created_at_ns, generation_id, directory, _generation in eligible: shutil.rmtree(directory) - removed.append(generation_id) + reclaimed.append(generation_id) # The retired-* markers only point at superseded generations, so they # follow the same retention -- otherwise they accumulate as dangling @@ -809,19 +943,56 @@ def prune_superseded_generations(self, *, keep: int = SUPERSEDED_GENERATION_RETE reverse=True, ) pruned_markers = 0 - for marker in markers[keep:]: + for marker in markers[SUPERSEDED_GENERATION_RETENTION:]: shutil.rmtree(marker) pruned_markers += 1 - if removed or pruned_markers: - # Markers count toward the fsync gate too: an archive's first - # marker is pruned a promotion before any gen-* becomes prunable, - # so gating on `removed` alone skipped the durability barrier - # exactly when only markers had gone. + if reclaimed or pruned_markers: _fsync_directory(self.generations_root) - if removed: - logger.info("pruned %d superseded index generation(s): %s", len(removed), ", ".join(removed)) - return removed + completed = GenerationRetentionReceipt( + promoted_generation_id=receipt.promoted_generation_id, + promoted_at_ns=receipt.promoted_at_ns, + retention_boundary=receipt.retention_boundary, + automatic=True, + records=tuple( + record + if record.generation_id not in reclaimed + else GenerationRetentionRecord( + generation_id=record.generation_id, + generation_owner_id=record.generation_owner_id, + retention_owner_id=record.retention_owner_id, + state=GenerationRetentionState.RECLAIMED, + ) + for record in receipt.records + ), + eligible_generation_ids=receipt.eligible_generation_ids, + reclaimed_marker_count=pruned_markers, + ) + self._write_retention_receipt(completed) + self._prune_retention_receipts(current_generation_id=completed.promoted_generation_id) + if reclaimed: + logger.info("reclaimed %d superseded index generation(s): %s", len(reclaimed), ", ".join(reclaimed)) + return completed + + def load_retention_receipt(self, promoted_generation_id: str) -> GenerationRetentionReceipt: + payload = json.loads(self._retention_receipt_path(promoted_generation_id).read_text(encoding="utf-8")) + return GenerationRetentionReceipt( + promoted_generation_id=str(payload["promoted_generation_id"]), + promoted_at_ns=int(payload.get("promoted_at_ns", 0)), + retention_boundary=int(payload["retention_boundary"]), + automatic=bool(payload["automatic"]), + records=tuple( + GenerationRetentionRecord( + generation_id=str(record["generation_id"]), + generation_owner_id=str(record["generation_owner_id"]), + retention_owner_id=str(record["retention_owner_id"]), + state=GenerationRetentionState(str(record["state"])), + ) + for record in payload["records"] + ), + eligible_generation_ids=tuple(str(generation_id) for generation_id in payload["eligible_generation_ids"]), + reclaimed_marker_count=int(payload.get("reclaimed_marker_count", 0)), + ) def recover_promotion(self, generation_id: str) -> IndexGeneration: """Reconcile an incomplete promotion without trusting the pointer alone. @@ -855,8 +1026,21 @@ def complete_promotion_recovery(self, generation_id: str) -> IndexGeneration: raise RuntimeError("cannot complete promotion recovery without an active index pointer") if pointer.resolve(strict=True) != Path(generation.index_path).resolve(strict=True): raise RuntimeError("cannot complete promotion recovery for a non-active generation") - recovered = IndexGeneration(**{**asdict(generation), "state": "active"}) + self._validate_retention_ownership() + recovered = IndexGeneration( + **{ + **asdict(generation), + "state": "active", + "promoted_at_ns": self._next_lifecycle_timestamp_ns(), + "retention_owner_id": generation.generation_id, + "retention_state": GenerationRetentionState.ACTIVE.value, + } + ) self._write(recovered) + try: + self._collect_superseded_generations(recovered) + except OSError: + logger.warning("index generation retention collection failed after recovered promotion", exc_info=True) return recovered def discard_if_inactive(self, generation: IndexGeneration) -> bool: @@ -871,6 +1055,42 @@ def discard_if_inactive(self, generation: IndexGeneration) -> bool: def _metadata_path(self, generation_id: str) -> Path: return self.generations_root / generation_id / "generation.json" + def _retention_receipt_path(self, promoted_generation_id: str) -> Path: + return self.generations_root / _RETENTION_RECEIPTS_DIRNAME / f"{promoted_generation_id}.json" + + def _generation_id_for_active_target(self, pointer: Path) -> str | None: + """Find the generation currently exposed by ``pointer``, if any.""" + if not (pointer.exists() or pointer.is_symlink()): + return None + try: + active_target = pointer.resolve(strict=True) + except OSError: + return None + for metadata_path in sorted(self.generations_root.glob("gen-*/generation.json")): + try: + generation = IndexGeneration(**json.loads(metadata_path.read_text(encoding="utf-8"))) + if generation.state == "active" and Path(generation.index_path).resolve(strict=True) == active_target: + return generation.generation_id + except (OSError, ValueError, TypeError): + continue + return None + + def _next_lifecycle_timestamp_ns(self) -> int: + """Return a persisted lifecycle timestamp that never moves backwards. + + Rebuild ownership serializes production promotion. Reading the small + bounded generation set here also keeps a fresh store instance monotonic + after a process restart or a coarse/frozen wall clock in a test. + """ + latest = 0 + for metadata_path in self.generations_root.glob("gen-*/generation.json"): + try: + generation = IndexGeneration(**json.loads(metadata_path.read_text(encoding="utf-8"))) + except (OSError, ValueError, TypeError): + continue + latest = max(latest, _generation_lifecycle_recency_ns(generation)) + return max(time.time_ns(), latest + 1) + def _transaction_path(self, operation_id: str) -> Path: return self.transactions_root / f"{operation_id}.json" @@ -881,6 +1101,33 @@ def _write(self, generation: IndexGeneration) -> None: os.replace(temporary, path) _fsync_directory(path.parent) + def _write_retention_receipt(self, receipt: GenerationRetentionReceipt) -> None: + path = self._retention_receipt_path(receipt.promoted_generation_id) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".json.tmp") + temporary.write_text(json.dumps(asdict(receipt), indent=2, sort_keys=True), encoding="utf-8") + os.replace(temporary, path) + _fsync_directory(path.parent) + + def _prune_retention_receipts(self, *, current_generation_id: str) -> None: + """Bound receipt history without deleting current or unreadable proof.""" + receipts_root = self.generations_root / _RETENTION_RECEIPTS_DIRNAME + candidates: list[tuple[int, str, Path]] = [] + for receipt_path in receipts_root.glob("*.json"): + if receipt_path == self._retention_receipt_path(current_generation_id): + continue + try: + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + promoted_at_ns = int(payload.get("promoted_at_ns", 0)) + except (OSError, ValueError, TypeError): + continue # malformed evidence remains visible for investigation + candidates.append((promoted_at_ns, receipt_path.name, receipt_path)) + candidates.sort(reverse=True) + for _promoted_at_ns, _name, receipt_path in candidates[RETENTION_RECEIPT_HISTORY - 1 :]: + receipt_path.unlink() + if len(candidates) >= RETENTION_RECEIPT_HISTORY: + _fsync_directory(receipts_root) + def source_revision_snapshot(archive_root: Path) -> str: """Hash the full mutable raw-session state after a rebuild replay.""" @@ -995,6 +1242,14 @@ def _checkpoint_truncate(path: Path, *, label: str) -> None: raise RuntimeError(f"{label} WAL checkpoint failed: {checkpoint!r}") +def _generation_creation_recency_ns(generation: IndexGeneration) -> int: + return generation.created_at_ns or generation.created_at_ms * 1_000_000 + + +def _generation_lifecycle_recency_ns(generation: IndexGeneration) -> int: + return generation.promoted_at_ns or _generation_creation_recency_ns(generation) + + def _fsync_directory(path: Path) -> None: fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) try: diff --git a/tests/unit/storage/test_index_generation.py b/tests/unit/storage/test_index_generation.py index 5dbe692fd1..1985683934 100644 --- a/tests/unit/storage/test_index_generation.py +++ b/tests/unit/storage/test_index_generation.py @@ -13,6 +13,7 @@ from polylogue.storage.archive_identity import ArchiveLocation from polylogue.storage.index_generation import ( + RETENTION_RECEIPT_HISTORY, ActiveWriterLease, IndexGenerationStore, RebuildLease, @@ -250,6 +251,75 @@ def test_recover_promotion_after_pointer_swap_does_not_mark_active(tmp_path: Pat assert completed.state == "active" +def test_recovered_promotion_records_automatic_retention_and_reclamation( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Recovery completion must use the same retention lifecycle as promotion. + + Anti-vacuity: this exercises the production crash-recovery seam after a + pointer swap. All three generations share a millisecond, while their UUID + text is deliberately reverse-chronological. Removing recovery's retention + collection leaves the third generation without a receipt; ordering by the + old ``(created_at_ms, generation_id)`` tuple retains the wrong rollback + target. + """ + _archive(tmp_path) + store = IndexGenerationStore.for_archive_root(tmp_path) + monkeypatch.setattr("polylogue.storage.index_generation.time.time_ns", lambda: 1_000_000_000) + uuid_hexes = iter(("ffffffff", "11111111", "22222222", "00000000", "33333333")) + monkeypatch.setattr( + "polylogue.storage.index_generation.uuid.uuid4", + lambda: type("DeterministicUuid", (), {"hex": next(uuid_hexes)})(), + ) + first = store.create(owner_id="build-1", source_snapshot="snapshot-1") + store.promote(first) + + recovered_generations = [] + for index in (2, 3): + generation = store.create(owner_id=f"build-{index}", source_snapshot=f"snapshot-{index}") + store._write(replace(generation, state="promoting")) + store.active_pointer.unlink() + store.active_pointer.symlink_to(generation.index_path) + recovered_generations.append(store.complete_promotion_recovery(generation.generation_id)) + + receipt = store.load_retention_receipt(recovered_generations[-1].generation_id) + + assert receipt.states_by_generation_id == { + recovered_generations[-1].generation_id: "active", + recovered_generations[-2].generation_id: "retained", + first.generation_id: "reclaimed", + } + assert Path(recovered_generations[-2].index_path).exists() + assert not Path(first.index_path).parent.exists() + assert { + first.created_at_ms, + recovered_generations[0].created_at_ms, + recovered_generations[1].created_at_ms, + } == {1_000} + + +def test_promotion_bounds_retention_receipt_history(tmp_path: Path) -> None: + """Receipt evidence is automatic, but its bounded history cannot grow forever. + + Anti-vacuity: this performs four real promotions, then inspects the + production receipt directory. Removing receipt pruning leaves all four + receipt files behind instead of the active and immediately prior proofs. + """ + _archive(tmp_path) + store = IndexGenerationStore.for_archive_root(tmp_path) + + promoted = [] + for index in range(4): + generation = store.create(owner_id=f"build-{index}", source_snapshot=f"snapshot-{index}") + store.promote(generation) + promoted.append(generation) + + receipts = {path.stem for path in (store.generations_root / "retention-receipts").glob("*.json")} + + assert RETENTION_RECEIPT_HISTORY == 2 + assert receipts == {promoted[-1].generation_id, promoted[-2].generation_id} + + def test_archive_store_init_failure_releases_writer_lease(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) monkeypatch.setattr( @@ -458,33 +528,100 @@ def test_promotion_prunes_superseded_generations(tmp_path: Path) -> None: assert len(list(store.generations_root.glob("retired-*"))) == 1 -def test_pruning_never_removes_the_active_generation(tmp_path: Path) -> None: - """Retention of zero still must not delete what the pointer resolves to.""" +def test_promotion_records_automatic_retention_and_reclamation(tmp_path: Path) -> None: + """The real blue-green seam retains one rollback generation, then records GC. + + Anti-vacuity: this calls ``IndexGenerationStore.promote`` against actual + SQLite index generations. Removing promotion's retention lifecycle call + leaves the prior generation's metadata untouched and no durable receipt, + so this test fails instead of merely checking a test-local planner. + """ _archive(tmp_path) store = IndexGenerationStore.for_archive_root(tmp_path) - generation = store.create(owner_id="operator", source_snapshot="snapshot-a") - store.promote(generation) - removed = store.prune_superseded_generations(keep=0) + promoted = [] + for index in range(3): + generation = store.create(owner_id=f"build-{index}", source_snapshot=f"snapshot-{index}") + store.promote(generation) + promoted.append(generation) + + receipt = store.load_retention_receipt(promoted[-1].generation_id) + + assert receipt.automatic is True + assert receipt.retention_boundary == 1 + assert receipt.states_by_generation_id == { + promoted[-1].generation_id: "active", + promoted[-2].generation_id: "retained", + promoted[-3].generation_id: "reclaimed", + } + assert receipt.eligible_generation_ids == (promoted[-3].generation_id,) + assert receipt.owner_by_generation_id[promoted[-2].generation_id] == promoted[-1].generation_id + assert receipt.owner_by_generation_id[promoted[-3].generation_id] == promoted[-1].generation_id + assert Path(promoted[-2].index_path).exists(), "the rollback generation was reclaimed before its boundary" + assert not Path(promoted[-3].index_path).parent.exists(), "eligible generation was not reclaimed automatically" + + +def test_promotion_retains_actual_predecessor_when_generation_ids_reverse( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Rollback retention follows the preceding pointer target, never UUID text. + + Anti-vacuity: this drives three production blue-green promotions with all + generation timestamps in one millisecond. The first ID sorts after the + actual predecessor, so the previous ``(created_at_ms, generation_id)`` + ordering retains the wrong generation after the third swap. + """ + _archive(tmp_path) + store = IndexGenerationStore.for_archive_root(tmp_path) + monkeypatch.setattr("polylogue.storage.index_generation.time.time", lambda: 1.0) + monkeypatch.setattr("polylogue.storage.index_generation.time.time_ns", lambda: 1_000_000_000) + uuid_hexes = iter( + ("ffffffff", "11111111", "22222222", "00000000", "33333333", "44444444", "55555555", "66666666", "77777777") + ) + monkeypatch.setattr( + "polylogue.storage.index_generation.uuid.uuid4", + lambda: type("DeterministicUuid", (), {"hex": next(uuid_hexes)})(), + ) + + promoted = [] + for index in range(3): + generation = store.create(owner_id=f"build-{index}", source_snapshot=f"snapshot-{index}") + store.promote(generation) + promoted.append(generation) - assert generation.generation_id not in removed - assert Path(generation.index_path).exists() - assert Path(store.active_pointer).resolve(strict=True) == Path(generation.index_path).resolve() + receipt = store.load_retention_receipt(promoted[-1].generation_id) + assert {generation.created_at_ms for generation in promoted} == {1_000} + assert receipt.states_by_generation_id == { + promoted[-1].generation_id: "active", + promoted[-2].generation_id: "retained", + promoted[-3].generation_id: "reclaimed", + } -def test_pruning_retains_everything_when_the_active_pointer_is_unresolvable(tmp_path: Path) -> None: - """Fail closed: if the thing that says what is live is missing, delete nothing.""" + +def test_promotion_refuses_ownerless_predecessor_before_pointer_swap(tmp_path: Path) -> None: + """An ownerless predecessor cannot become an unaccountable GC candidate. + + The mutation that makes this fail is removing the promotion-time ownership + preflight. The old promotion path accepted this corrupted predecessor, + changed the active pointer, and left later GC unable to prove who owned + the superseded generation. + """ _archive(tmp_path) store = IndexGenerationStore.for_archive_root(tmp_path) - first = store.create(owner_id="operator", source_snapshot="snapshot-a") - store.promote(first) - second = store.create(owner_id="operator", source_snapshot="snapshot-b") - store.promote(second) - store.active_pointer.unlink() - - assert store.prune_superseded_generations(keep=0) == [] - assert Path(first.index_path).exists() - assert Path(second.index_path).exists() + predecessor = store.create(owner_id="first-build", source_snapshot="snapshot-a") + store.promote(predecessor) + predecessor_metadata = Path(predecessor.index_path).with_name("generation.json") + payload = json.loads(predecessor_metadata.read_text(encoding="utf-8")) + payload["owner_id"] = "" + predecessor_metadata.write_text(json.dumps(payload), encoding="utf-8") + candidate = store.create(owner_id="second-build", source_snapshot="snapshot-b") + + with pytest.raises(RuntimeError, match="retention ownership"): + store.promote(candidate) + + assert Path(store.active_pointer).resolve(strict=True) == Path(predecessor.index_path).resolve() + assert store.load(candidate.generation_id).state == "inactive" def test_pruning_never_removes_a_never_promoted_rebuild_candidate(tmp_path: Path) -> None: