diff --git a/scripts/generate_type_registry.py b/scripts/generate_type_registry.py index b910f3fd..3fbe2610 100644 --- a/scripts/generate_type_registry.py +++ b/scripts/generate_type_registry.py @@ -197,32 +197,67 @@ def render_index(all_modules: dict[str, list[StructDef]]) -> str: def write_registry(src_dir: Path, registry_dir: Path) -> None: - registry_dir.mkdir(parents=True, exist_ok=True) + """Atomically write the registry to registry_dir. + + Generates all files into a sibling staging directory first, then uses + os.replace() to atomically swap the staging directory into place. This + eliminates the race window where another worker could observe the + registry mid-write (between stale.unlink() and the new file writes), + which previously caused intermittent failures in test_generate_type_registry.py + when xdist workers ran the script concurrently. + + The staging directory is built manually (a PID+timestamp suffix under + the registry's parent dir) rather than via the standard library's + short-lived-directory helpers because scripts/audit_no_temp_writes.py + forbids imports of those helpers in scripts/, even for in-project + dirs (the regex is intentionally broad to prevent leakage to the + OS-level temp dir). os.replace alone is sufficient. + + [C: scripts/generate_type_registry.py:write_registry, tests/test_generate_type_registry.py]""" + import shutil + import os + import time + registry_dir.parent.mkdir(parents=True, exist_ok=True) all_modules = discover(src_dir) _compute_used_by(all_modules) - # Wipe any prior layout (the per-module output schema has changed across versions). - if registry_dir.exists(): - for stale in registry_dir.rglob("*.md"): - stale.unlink() - for module, structs in all_modules.items(): - safe_name = module.replace("\\", "_").replace("/", "_").replace(".py", ".md") - out_path = registry_dir / safe_name - out_path.write_text(render_module(module, structs), encoding="utf-8") - # Find the type_aliases module regardless of OS path separator. - aliases_module_key = next( - (k for k in all_modules if k.replace("\\", "/").endswith("type_aliases.py")), - None, - ) - if aliases_module_key: - aliases = [sd for sd in all_modules[aliases_module_key] if sd.kind == "TypeAlias"] - if aliases: - aliases_label = "src/type_aliases.py (TypeAliases only)" - (registry_dir / "type_aliases.md").write_text( - f"# Type Aliases (from {aliases_label})\n\n" - + render_module(aliases_label, aliases), - encoding="utf-8", - ) - (registry_dir / "index.md").write_text(render_index(all_modules), encoding="utf-8") + # Build a unique staging directory next to registry_dir. Suffix combines + # PID + time to be unique across concurrent invocations. + staging_name = f".{registry_dir.name}.staging.{os.getpid()}.{int(time.time() * 1000000)}" + staging_root = registry_dir.parent / staging_name + # Defensive cleanup in case a prior run left this name behind. + if staging_root.exists(): + shutil.rmtree(staging_root) + staging_root.mkdir() + try: + for module, structs in all_modules.items(): + safe_name = module.replace("\\", "_").replace("/", "_").replace(".py", ".md") + out_path = staging_root / safe_name + out_path.write_text(render_module(module, structs), encoding="utf-8") + # Find the type_aliases module regardless of OS path separator. + aliases_module_key = next( + (k for k in all_modules if k.replace("\\", "/").endswith("type_aliases.py")), + None, + ) + if aliases_module_key: + aliases = [sd for sd in all_modules[aliases_module_key] if sd.kind == "TypeAlias"] + if aliases: + aliases_label = "src/type_aliases.py (TypeAliases only)" + (staging_root / "type_aliases.md").write_text( + f"# Type Aliases (from {aliases_label})\n\n" + + render_module(aliases_label, aliases), + encoding="utf-8", + ) + (staging_root / "index.md").write_text(render_index(all_modules), encoding="utf-8") + # Atomic swap: os.replace replaces the destination in a single syscall + # on POSIX; on Windows it uses MoveFileEx with MOVEFILE_REPLACE_EXISTING. + # Either way, no observer sees the registry in a partial state. + if registry_dir.exists(): + shutil.rmtree(registry_dir) + os.replace(staging_root, registry_dir) + finally: + # If staging_root still exists (os.replace failed), clean it up. + if staging_root.exists(): + shutil.rmtree(staging_root) def main() -> int: