object_integrity.py
python
sha256:1ccb8409daa5aabe577bd26d11b56ed12f3376d64011d0e75a247e81211a66ee
docs(mwp4/phase5): tick Phase 5 checkboxes, close musehub#109
Sonnet 4.6
20 days ago
| 1 | """Periodic object-store integrity scanner and soft-delete reaper for MuseHub. |
| 2 | |
| 3 | Object integrity |
| 4 | ---------------- |
| 5 | Content-addressed objects are immutable once written. This module provides |
| 6 | ``scan_object_integrity(session, backend, sample_size)`` which: |
| 7 | |
| 8 | 1. Draws a random sample of ``MusehubObject`` rows (default 200). |
| 9 | 2. Re-reads each object from the storage backend. |
| 10 | 3. Re-computes SHA-256(content) and compares it with the declared object_id. |
| 11 | 4. Returns an ``ObjectIntegrityScanResult`` listing any mismatches. |
| 12 | |
| 13 | Run this on a schedule (e.g. nightly cron or health-check endpoint) — not on |
| 14 | the hot request path. |
| 15 | |
| 16 | Soft-delete reaper |
| 17 | ------------------ |
| 18 | ``reap_deleted_objects(session, backend, retention_days)`` hard-deletes objects |
| 19 | whose ``deleted_at`` is older than ``retention_days`` (default ``object_retention_days`` |
| 20 | from settings). It removes the backing file/S3 object *and* the DB row. |
| 21 | |
| 22 | Call only during maintenance windows. |
| 23 | """ |
| 24 | |
| 25 | import logging |
| 26 | from dataclasses import dataclass, field |
| 27 | from datetime import datetime, timedelta, timezone |
| 28 | |
| 29 | from sqlalchemy import delete, func, select |
| 30 | from sqlalchemy.ext.asyncio import AsyncSession |
| 31 | |
| 32 | from muse.core.types import blob_id, split_id |
| 33 | from musehub.config import settings |
| 34 | from musehub.db.musehub_repo_models import MusehubObject |
| 35 | from musehub.storage.backends import StorageBackend |
| 36 | |
| 37 | logger = logging.getLogger(__name__) |
| 38 | |
| 39 | _DEFAULT_SAMPLE_SIZE = 200 |
| 40 | |
| 41 | @dataclass |
| 42 | class ObjectIntegrityMismatch: |
| 43 | """One corrupted or missing object found during a scan pass.""" |
| 44 | object_id: str |
| 45 | repo_id: str |
| 46 | reason: str # "missing" | "hash_mismatch" |
| 47 | |
| 48 | @dataclass |
| 49 | class ObjectIntegrityScanResult: |
| 50 | """Summary of one integrity scan pass.""" |
| 51 | sampled: int = 0 |
| 52 | mismatches: list[ObjectIntegrityMismatch] = field(default_factory=list) |
| 53 | |
| 54 | @property |
| 55 | def ok(self) -> bool: |
| 56 | return len(self.mismatches) == 0 |
| 57 | |
| 58 | @property |
| 59 | def mismatch_count(self) -> int: |
| 60 | return len(self.mismatches) |
| 61 | |
| 62 | async def scan_object_integrity( |
| 63 | session: AsyncSession, |
| 64 | backend: StorageBackend, |
| 65 | sample_size: int = _DEFAULT_SAMPLE_SIZE, |
| 66 | ) -> ObjectIntegrityScanResult: |
| 67 | """Sample up to ``sample_size`` live objects and verify their SHA-256 hashes. |
| 68 | |
| 69 | Read-only — does not modify the database or storage. |
| 70 | """ |
| 71 | rows = ( |
| 72 | await session.execute( |
| 73 | select(MusehubObject.object_id) |
| 74 | .where(MusehubObject.deleted_at.is_(None)) |
| 75 | .order_by(func.random()) |
| 76 | .limit(sample_size) |
| 77 | ) |
| 78 | ).scalars().all() |
| 79 | |
| 80 | result = ObjectIntegrityScanResult(sampled=len(rows)) |
| 81 | |
| 82 | for object_id in rows: |
| 83 | content = await backend.get(object_id) |
| 84 | if content is None: |
| 85 | result.mismatches.append( |
| 86 | ObjectIntegrityMismatch( |
| 87 | object_id=object_id, |
| 88 | repo_id="", |
| 89 | reason="missing", |
| 90 | ) |
| 91 | ) |
| 92 | logger.warning("integrity: object missing from storage: %s", object_id) |
| 93 | continue |
| 94 | |
| 95 | if object_id.startswith("sha256:"): |
| 96 | _, declared = split_id(object_id) |
| 97 | _, actual = split_id(blob_id(content)) |
| 98 | if actual != declared: |
| 99 | result.mismatches.append( |
| 100 | ObjectIntegrityMismatch( |
| 101 | object_id=object_id, |
| 102 | repo_id="", |
| 103 | reason="hash_mismatch", |
| 104 | ) |
| 105 | ) |
| 106 | logger.error( |
| 107 | "integrity: hash mismatch for %s: declared=%s actual=%s", |
| 108 | object_id, declared, actual, |
| 109 | ) |
| 110 | |
| 111 | return result |
| 112 | |
| 113 | async def soft_delete_object( |
| 114 | session: AsyncSession, |
| 115 | object_id: str, |
| 116 | ) -> bool: |
| 117 | """Mark an object as logically deleted by setting deleted_at to now. |
| 118 | |
| 119 | Returns True if the row was found and updated, False if not found. |
| 120 | Does NOT remove the backing file — that is deferred to ``reap_deleted_objects``. |
| 121 | """ |
| 122 | row = await session.get(MusehubObject, object_id) |
| 123 | if row is None: |
| 124 | return False |
| 125 | if row.deleted_at is None: |
| 126 | row.deleted_at = datetime.now(tz=timezone.utc) |
| 127 | await session.flush() |
| 128 | return True |
| 129 | |
| 130 | async def reap_deleted_objects( |
| 131 | session: AsyncSession, |
| 132 | backend: StorageBackend, |
| 133 | retention_days: int | None = None, |
| 134 | ) -> int: |
| 135 | """Hard-delete objects past the retention window. |
| 136 | |
| 137 | For each object whose ``deleted_at`` is older than ``retention_days``: |
| 138 | 1. Remove the backing file / S3 object from storage. |
| 139 | 2. Delete the DB row. |
| 140 | |
| 141 | Returns the number of objects permanently removed. |
| 142 | Commits the transaction before returning. |
| 143 | |
| 144 | Call only during a maintenance window — not on the hot request path. |
| 145 | """ |
| 146 | days = retention_days if retention_days is not None else settings.object_retention_days |
| 147 | cutoff = datetime.now(tz=timezone.utc) - timedelta(days=days) |
| 148 | |
| 149 | rows = ( |
| 150 | await session.execute( |
| 151 | select(MusehubObject.object_id) |
| 152 | .where( |
| 153 | MusehubObject.deleted_at.isnot(None), |
| 154 | MusehubObject.deleted_at < cutoff, |
| 155 | ) |
| 156 | ) |
| 157 | ).scalars().all() |
| 158 | |
| 159 | reaped = 0 |
| 160 | for object_id in rows: |
| 161 | try: |
| 162 | await backend.delete(object_id) |
| 163 | except Exception as exc: |
| 164 | logger.warning("reap: failed to delete storage for %s: %s", object_id, exc) |
| 165 | |
| 166 | await session.execute( |
| 167 | delete(MusehubObject).where(MusehubObject.object_id == object_id) |
| 168 | ) |
| 169 | reaped += 1 |
| 170 | |
| 171 | await session.commit() |
| 172 | logger.info("reap: hard-deleted %d object(s) past %d-day retention window", reaped, days) |
| 173 | return reaped |
File History
2 commits
sha256:1ccb8409daa5aabe577bd26d11b56ed12f3376d64011d0e75a247e81211a66ee
docs(mwp4/phase5): tick Phase 5 checkboxes, close musehub#109
Sonnet 4.6
20 days ago
sha256:2c523da45351334b5c4dbefed4dc3dd553b3faa8737a4e6caf301e5dc82141be
test(mwp4): Phase 0 RED reproduction tests for RC-4 ordering race
Sonnet 4.6
20 days ago