backfill_move_to_address.py
python
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago
| 1 | """Backfill to_address into DELETE rows that are move-source counterparts. |
| 2 | |
| 3 | When backfill_history_from_snapshots previously ran, move-source paths were |
| 4 | not recorded at all (the delete was suppressed). Now they are recorded with |
| 5 | op_payload.to_address pointing to the new path. |
| 6 | |
| 7 | This script finds existing DELETE rows that are missing to_address and |
| 8 | re-derives it by matching the corresponding MOVE row in the same commit |
| 9 | (same repo, same commit_id, move.op_payload.from_address == delete.address). |
| 10 | |
| 11 | Usage: |
| 12 | docker exec musehub python3 /app/deploy/backfill_move_to_address.py --dry-run |
| 13 | docker exec musehub python3 /app/deploy/backfill_move_to_address.py |
| 14 | docker exec musehub python3 /app/deploy/backfill_move_to_address.py --repo-id <id> |
| 15 | """ |
| 16 | from __future__ import annotations |
| 17 | |
| 18 | import argparse |
| 19 | import asyncio |
| 20 | import time |
| 21 | |
| 22 | import sqlalchemy as sa |
| 23 | from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine |
| 24 | from sqlalchemy.orm import sessionmaker |
| 25 | |
| 26 | from musehub.db.database import get_database_url |
| 27 | from musehub.db import musehub_models as db |
| 28 | |
| 29 | |
| 30 | async def run(repo_id: str | None, dry_run: bool) -> None: |
| 31 | engine = create_async_engine(get_database_url(), echo=False) |
| 32 | async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| 33 | |
| 34 | async with async_session() as session: |
| 35 | t0 = time.monotonic() |
| 36 | count = await backfill_move_to_address(session, repo_id=repo_id, dry_run=dry_run) |
| 37 | if not dry_run: |
| 38 | await session.commit() |
| 39 | elapsed = time.monotonic() - t0 |
| 40 | |
| 41 | verb = "Would update" if dry_run else "Updated" |
| 42 | scope = f" for repo {repo_id}" if repo_id else " across all repos" |
| 43 | print(f"{verb} {count} DELETE rows with to_address{scope} in {elapsed:.1f}s") |
| 44 | |
| 45 | |
| 46 | async def backfill_move_to_address( |
| 47 | session: AsyncSession, |
| 48 | repo_id: str | None = None, |
| 49 | *, |
| 50 | dry_run: bool = False, |
| 51 | ) -> int: |
| 52 | """For each MOVE row, find its corresponding DELETE row (same commit, same |
| 53 | repo, delete.address == move.op_payload.from_address) and write |
| 54 | to_address into the DELETE's op_payload. |
| 55 | |
| 56 | Skips DELETE rows that already have to_address set. |
| 57 | Also creates DELETE rows that are completely missing (old backfill run |
| 58 | suppressed them entirely). |
| 59 | """ |
| 60 | she = db.MusehubSymbolHistoryEntry |
| 61 | |
| 62 | # Load all rows that carry from_address in op_payload — covers both |
| 63 | # snapshot-diff MOVE rows and structured_delta PATCH/move rows. |
| 64 | ref_q = sa.select(she).where( |
| 65 | she.op_payload["from_address"].as_string() != None # noqa: E711 |
| 66 | ) |
| 67 | if repo_id is not None: |
| 68 | ref_q = ref_q.where(she.repo_id == repo_id) |
| 69 | ref_rows = (await session.execute(ref_q)).scalars().all() |
| 70 | |
| 71 | if not ref_rows: |
| 72 | return 0 |
| 73 | |
| 74 | # Build (repo_id, commit_id, from_address) → new_address map |
| 75 | move_map: dict[tuple[str, str, str], str] = {} |
| 76 | for row in ref_rows: |
| 77 | from_addr = (row.op_payload or {}).get("from_address") |
| 78 | if from_addr: |
| 79 | move_map[(row.repo_id, row.commit_id, from_addr)] = row.address |
| 80 | |
| 81 | if not move_map: |
| 82 | return 0 |
| 83 | |
| 84 | # Load existing DELETE rows for those (repo_id, commit_id, address) keys |
| 85 | # to determine which need updating vs. creating. |
| 86 | from_addrs = list({k[2] for k in move_map}) |
| 87 | existing_q = sa.select(she).where( |
| 88 | she.op == "delete", |
| 89 | she.address.in_(from_addrs), |
| 90 | ) |
| 91 | if repo_id is not None: |
| 92 | existing_q = existing_q.where(she.repo_id == repo_id) |
| 93 | existing_rows = (await session.execute(existing_q)).scalars().all() |
| 94 | |
| 95 | existing_map: dict[tuple[str, str, str], she] = { # type: ignore[valid-type] |
| 96 | (r.repo_id, r.commit_id, r.address): r for r in existing_rows |
| 97 | } |
| 98 | |
| 99 | # Also load commit metadata for creating missing rows |
| 100 | commit_ids = list({k[1] for k in move_map}) |
| 101 | commit_q = sa.select(db.MusehubCommit).where( |
| 102 | db.MusehubCommit.commit_id.in_(commit_ids) |
| 103 | ) |
| 104 | if repo_id is not None: |
| 105 | commit_q = commit_q.where(db.MusehubCommit.repo_id == repo_id) |
| 106 | commit_map = {c.commit_id: c for c in (await session.execute(commit_q)).scalars().all()} |
| 107 | |
| 108 | updated = 0 |
| 109 | for (r_id, c_id, from_addr), to_addr in move_map.items(): |
| 110 | key = (r_id, c_id, from_addr) |
| 111 | existing = existing_map.get(key) |
| 112 | |
| 113 | if existing is not None: |
| 114 | if (existing.op_payload or {}).get("to_address"): |
| 115 | continue # already set |
| 116 | if not dry_run: |
| 117 | import json as _json |
| 118 | full_payload = dict(existing.op_payload or {}) |
| 119 | full_payload["to_address"] = to_addr |
| 120 | await session.execute( |
| 121 | sa.text( |
| 122 | "UPDATE musehub_symbol_history_entries" |
| 123 | " SET op_payload = CAST(:payload AS json)" |
| 124 | " WHERE repo_id = :repo_id" |
| 125 | " AND commit_id = :commit_id" |
| 126 | " AND address = :address" |
| 127 | " AND op = 'delete'" |
| 128 | ), |
| 129 | { |
| 130 | "payload": _json.dumps(full_payload), |
| 131 | "repo_id": r_id, |
| 132 | "commit_id": c_id, |
| 133 | "address": from_addr, |
| 134 | }, |
| 135 | ) |
| 136 | updated += 1 |
| 137 | else: |
| 138 | # Row missing entirely — create it |
| 139 | commit = commit_map.get(c_id) |
| 140 | if commit is None: |
| 141 | continue |
| 142 | payload = { |
| 143 | "inferred_from": "snapshot_diff", |
| 144 | "to_address": to_addr, |
| 145 | } |
| 146 | if not dry_run: |
| 147 | session.add(db.MusehubSymbolHistoryEntry( |
| 148 | repo_id=r_id, |
| 149 | address=from_addr, |
| 150 | commit_id=c_id, |
| 151 | op="delete", |
| 152 | op_payload=payload, |
| 153 | content_id=None, |
| 154 | committed_at=commit.timestamp, |
| 155 | author=commit.author or "", |
| 156 | )) |
| 157 | updated += 1 |
| 158 | |
| 159 | return updated |
| 160 | |
| 161 | |
| 162 | def main() -> None: |
| 163 | parser = argparse.ArgumentParser(description=__doc__) |
| 164 | parser.add_argument("--dry-run", action="store_true") |
| 165 | parser.add_argument("--repo-id", default=None) |
| 166 | parser.add_argument("-q", "--quiet", action="store_true") |
| 167 | args = parser.parse_args() |
| 168 | |
| 169 | if args.quiet: |
| 170 | import logging |
| 171 | logging.disable(logging.CRITICAL) |
| 172 | |
| 173 | asyncio.run(run(args.repo_id, args.dry_run)) |
| 174 | |
| 175 | |
| 176 | if __name__ == "__main__": |
| 177 | main() |
File History
1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago