migrate_flat_to_per_repo.py
python
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago
| 1 | """Migrate legacy flat-store objects to the per-repo layout. |
| 2 | |
| 3 | Run in order: |
| 4 | |
| 5 | python3 migrate_flat_to_per_repo.py --dry-run # count, touch nothing |
| 6 | python3 migrate_flat_to_per_repo.py --migrate # copy legacy → per-repo (no deletes) |
| 7 | python3 migrate_flat_to_per_repo.py --verify # confirm every ref exists in per-repo |
| 8 | python3 migrate_flat_to_per_repo.py --prune # delete legacy files (only after verify passes) |
| 9 | |
| 10 | Each object is copied into EVERY repo that references it (per the musehub_object_refs table), |
| 11 | so the per-repo store is fully self-contained for each repo. |
| 12 | |
| 13 | Source resolution per object (in order): |
| 14 | 1. disk_path if non-empty and is a regular file |
| 15 | 2. storage_uri stripped of "local://" prefix |
| 16 | |
| 17 | Legacy source layouts covered: |
| 18 | /data/musehub/objects/objects/<64hex> flat store |
| 19 | /data/musehub/objects/<uuid>/<64hex> old UUID-per-repo store |
| 20 | /data/musehub/sha256_<hex> oldest single-file layout |
| 21 | |
| 22 | Target layout: |
| 23 | /data/repos/<owner>/<slug>/objects/sha256/<2hex>/<62hex> |
| 24 | |
| 25 | Every step is idempotent — safe to re-run. --migrate never deletes. |
| 26 | --prune refuses to run if --verify has not passed cleanly first. |
| 27 | """ |
| 28 | from __future__ import annotations |
| 29 | |
| 30 | import argparse |
| 31 | import asyncio |
| 32 | import os |
| 33 | import sys |
| 34 | from pathlib import Path |
| 35 | |
| 36 | sys.path.insert(0, "/app") |
| 37 | sys.path.insert(0, "/tmp/devpkgs") |
| 38 | |
| 39 | REPOS_DIR = Path("/data/repos") |
| 40 | |
| 41 | |
| 42 | async def load_rows(engine) -> list: |
| 43 | """Return ALL (object_id, disk_path, storage_uri, owner, slug) rows. |
| 44 | |
| 45 | One row per (object × repo) combination — objects referenced by N repos |
| 46 | produce N rows so the object is copied into every referencing repo's store. |
| 47 | Source fields (disk_path, storage_uri) are identical across all rows for |
| 48 | the same object_id. |
| 49 | """ |
| 50 | import sqlalchemy as sa |
| 51 | from sqlalchemy.ext.asyncio import async_sessionmaker |
| 52 | |
| 53 | sf = async_sessionmaker(engine, expire_on_commit=False) |
| 54 | async with sf() as s: |
| 55 | result = await s.execute(sa.text( |
| 56 | """ |
| 57 | SELECT |
| 58 | obj.object_id, |
| 59 | obj.disk_path, |
| 60 | obj.storage_uri, |
| 61 | r.owner, |
| 62 | r.slug |
| 63 | FROM musehub_objects obj |
| 64 | JOIN musehub_object_refs ref ON ref.object_id = obj.object_id |
| 65 | JOIN musehub_repos r ON r.repo_id = ref.repo_id |
| 66 | ORDER BY obj.object_id, r.owner, r.slug |
| 67 | """ |
| 68 | )) |
| 69 | return result.fetchall() |
| 70 | |
| 71 | |
| 72 | def resolve_src(disk_path: str, storage_uri: str | None) -> Path | None: |
| 73 | """Return the source Path for an object, or None if unresolvable.""" |
| 74 | if disk_path: |
| 75 | p = Path(disk_path) |
| 76 | if p.is_file(): |
| 77 | return p |
| 78 | if storage_uri and storage_uri.startswith("local://"): |
| 79 | p = Path(storage_uri[len("local://"):]) |
| 80 | if p.is_file(): |
| 81 | return p |
| 82 | return None |
| 83 | |
| 84 | |
| 85 | def target_path(object_id: str, owner: str, slug: str) -> Path: |
| 86 | from musehub.storage.backends import repo_root_for |
| 87 | from muse.core.paths import server_objects_dir |
| 88 | |
| 89 | hex_str = object_id[7:] # strip "sha256:" |
| 90 | repo_root = repo_root_for(owner, slug, repos_dir=str(REPOS_DIR)) |
| 91 | objects_dir = server_objects_dir(repo_root) |
| 92 | return objects_dir / "sha256" / hex_str[:2] / hex_str[2:] |
| 93 | |
| 94 | |
| 95 | async def cmd_dry_run(rows: list) -> None: |
| 96 | would_copy = 0 |
| 97 | already_present = 0 |
| 98 | source_missing = 0 |
| 99 | |
| 100 | for object_id, disk_path, storage_uri, owner, slug in rows: |
| 101 | if not object_id.startswith("sha256:"): |
| 102 | source_missing += 1 |
| 103 | continue |
| 104 | src = resolve_src(disk_path, storage_uri) |
| 105 | if src is None: |
| 106 | print(f" NO SOURCE: {object_id[:28]} ({owner}/{slug})") |
| 107 | source_missing += 1 |
| 108 | continue |
| 109 | tgt = target_path(object_id, owner, slug) |
| 110 | if tgt.exists(): |
| 111 | already_present += 1 |
| 112 | else: |
| 113 | would_copy += 1 |
| 114 | |
| 115 | print("[DRY RUN]") |
| 116 | print(f" would copy: {would_copy}") |
| 117 | print(f" already present: {already_present}") |
| 118 | print(f" source missing: {source_missing}") |
| 119 | |
| 120 | |
| 121 | async def cmd_migrate(rows: list) -> int: |
| 122 | copied = 0 |
| 123 | already_present = 0 |
| 124 | source_missing = 0 |
| 125 | errors = 0 |
| 126 | |
| 127 | for object_id, disk_path, storage_uri, owner, slug in rows: |
| 128 | if not object_id.startswith("sha256:"): |
| 129 | source_missing += 1 |
| 130 | continue |
| 131 | src = resolve_src(disk_path, storage_uri) |
| 132 | if src is None: |
| 133 | print(f" NO SOURCE: {object_id[:28]} ({owner}/{slug})") |
| 134 | source_missing += 1 |
| 135 | continue |
| 136 | tgt = target_path(object_id, owner, slug) |
| 137 | if tgt.exists(): |
| 138 | already_present += 1 |
| 139 | continue |
| 140 | try: |
| 141 | tgt.parent.mkdir(parents=True, exist_ok=True) |
| 142 | tmp = tgt.with_suffix(".tmp") |
| 143 | tmp.write_bytes(src.read_bytes()) |
| 144 | tmp.rename(tgt) |
| 145 | copied += 1 |
| 146 | except Exception as exc: |
| 147 | print(f" ERROR {object_id[:28]} ({owner}/{slug}): {exc}") |
| 148 | errors += 1 |
| 149 | |
| 150 | print("\n[MIGRATE]") |
| 151 | print(f" copied: {copied}") |
| 152 | print(f" already present: {already_present}") |
| 153 | print(f" source missing: {source_missing}") |
| 154 | print(f" errors: {errors}") |
| 155 | return errors |
| 156 | |
| 157 | |
| 158 | async def cmd_verify(rows: list) -> int: |
| 159 | ok = 0 |
| 160 | missing = 0 |
| 161 | |
| 162 | seen_missing: set[tuple[str, str, str]] = set() |
| 163 | for object_id, disk_path, storage_uri, owner, slug in rows: |
| 164 | if not object_id.startswith("sha256:"): |
| 165 | continue |
| 166 | tgt = target_path(object_id, owner, slug) |
| 167 | if tgt.exists(): |
| 168 | ok += 1 |
| 169 | else: |
| 170 | key = (object_id, owner, slug) |
| 171 | if key not in seen_missing: |
| 172 | print(f" MISSING: {object_id[:28]} ({owner}/{slug})") |
| 173 | seen_missing.add(key) |
| 174 | missing += 1 |
| 175 | |
| 176 | print("\n[VERIFY]") |
| 177 | print(f" ok: {ok}") |
| 178 | print(f" missing: {missing}") |
| 179 | if missing == 0: |
| 180 | print(" ✅ All objects present in per-repo store.") |
| 181 | else: |
| 182 | print(f" ❌ {missing} object(s) not in per-repo store — run --migrate first.") |
| 183 | return missing |
| 184 | |
| 185 | |
| 186 | async def cmd_prune(rows: list) -> None: |
| 187 | # Collect unique source paths across all rows for the same object |
| 188 | seen_src: set[Path] = set() |
| 189 | for object_id, disk_path, storage_uri, owner, slug in rows: |
| 190 | if not object_id.startswith("sha256:"): |
| 191 | continue |
| 192 | src = resolve_src(disk_path, storage_uri) |
| 193 | if src is not None: |
| 194 | seen_src.add(src) |
| 195 | |
| 196 | deleted = 0 |
| 197 | already_gone = 0 |
| 198 | errors = 0 |
| 199 | |
| 200 | for src in seen_src: |
| 201 | if not src.exists(): |
| 202 | already_gone += 1 |
| 203 | continue |
| 204 | try: |
| 205 | src.unlink() |
| 206 | deleted += 1 |
| 207 | except Exception as exc: |
| 208 | print(f" ERROR deleting {src}: {exc}") |
| 209 | errors += 1 |
| 210 | |
| 211 | print("\n[PRUNE]") |
| 212 | print(f" deleted: {deleted}") |
| 213 | print(f" already gone: {already_gone}") |
| 214 | print(f" errors: {errors}") |
| 215 | if errors == 0: |
| 216 | print("\n Legacy source files removed.") |
| 217 | print(" You can now: rm -rf /data/musehub/objects") |
| 218 | |
| 219 | |
| 220 | async def main(args: argparse.Namespace) -> None: |
| 221 | import sqlalchemy as sa |
| 222 | from sqlalchemy.ext.asyncio import create_async_engine |
| 223 | |
| 224 | db_url = os.environ["DATABASE_URL"] |
| 225 | engine = create_async_engine(db_url, echo=False) |
| 226 | |
| 227 | rows = await load_rows(engine) |
| 228 | print(f"Object×repo ref pairs: {len(rows)}\n") |
| 229 | |
| 230 | if args.dry_run: |
| 231 | await cmd_dry_run(rows) |
| 232 | elif args.migrate: |
| 233 | errors = await cmd_migrate(rows) |
| 234 | if errors: |
| 235 | sys.exit(1) |
| 236 | elif args.verify: |
| 237 | missing = await cmd_verify(rows) |
| 238 | if missing: |
| 239 | sys.exit(1) |
| 240 | elif args.prune: |
| 241 | missing = await cmd_verify(rows) |
| 242 | if missing: |
| 243 | print("\n❌ Aborting prune — verify failed. Run --migrate first.") |
| 244 | sys.exit(1) |
| 245 | await cmd_prune(rows) |
| 246 | |
| 247 | await engine.dispose() |
| 248 | |
| 249 | |
| 250 | def parse_args() -> argparse.Namespace: |
| 251 | p = argparse.ArgumentParser(description=__doc__, |
| 252 | formatter_class=argparse.RawDescriptionHelpFormatter) |
| 253 | g = p.add_mutually_exclusive_group(required=True) |
| 254 | g.add_argument("--dry-run", action="store_true", help="Count, touch nothing") |
| 255 | g.add_argument("--migrate", action="store_true", help="Copy legacy → per-repo (no deletes)") |
| 256 | g.add_argument("--verify", action="store_true", help="Confirm all refs exist in per-repo") |
| 257 | g.add_argument("--prune", action="store_true", |
| 258 | help="Delete legacy files (runs verify first, aborts if not clean)") |
| 259 | return p.parse_args() |
| 260 | |
| 261 | |
| 262 | if __name__ == "__main__": |
| 263 | asyncio.run(main(parse_args())) |
File History
1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago