"""Does a file rename work as an exclusive claim on Windows?

Licence: MIT. Copyright (c) 2026 QuantumMind AI LLC. See LICENSE.txt beside this file.

N processes race to rename each of M files to a name of their own
(task_007.txt -> task_007.txt.w0, task_007.txt.w1, ...). A process whose
rename call returns without an error believes it has claimed the file.

Three claim rules (mode):
  plain    a claim stands when the rename call returns without error
  verify   after a successful rename the process opens its new name for
           reading (Python open(path, "rb"), which does not share delete
           access) and closes it at once. ANY failed open means the claim
           is lost.
  retry    as verify, but the failed open is classified:
             FileNotFoundError        the file is not under our name: the
                                      claim is lost
             PermissionError          (what open() raises for a sharing
                                      violation, WinError 32: the file is
                                      there but another handle is open):
                                      retry for up to 1 second, then give up
             any other OSError        the claim is lost

Two timings (sync):
  barrier  all processes are released together before each file
  free     each process just loops over the files, no common release

After each round every file is counted once as:
  claimed once     exactly one process believes it has the file (the good case)
  claimed twice    two or more processes believe it (a double claim)
  claimed by nobody  no process believes it, but the file is there

It also counts why verify-opens failed (not found / permission denied / other).
A short demo (--no-demo to skip) shows two Windows behaviours directly.

Standard library only (ctypes for the demo). Uses a temporary folder of its
own and removes it. Usage:
  py -3.12 -B repro_rename_race.py [--procs 5] [--files 60] [--rounds 10]
It stops by itself and gives up after --max-seconds (default 110).
"""
import argparse
import multiprocessing as mp
import os
import platform
import shutil
import sys
import tempfile
import time

RETRY_SECONDS = 1.0


def try_open(dst, retry):
    """Return (ok, reason). reason in notfound / denied / other / ''."""
    deadline = time.time() + RETRY_SECONDS
    while True:
        try:
            with open(dst, "rb"):
                pass
            return True, ""
        except FileNotFoundError:
            return False, "notfound"
        except PermissionError:
            # open() reports a Windows sharing violation (WinError 32) as
            # PermissionError errno 13 with no winerror, so a sharing
            # violation and a plain access-denied look the same here.
            if retry and time.time() < deadline:
                time.sleep(0.005)
                continue
            return False, "denied"
        except OSError as e:
            return False, "other:%s/%s" % (e.errno, getattr(e, "winerror", None))


def worker(idx, mode, sync, src_dir, dst_dir, names, barrier, out_q):
    claimed = []
    fails = {}
    rename_errors = 0
    for name in names:
        if sync == "barrier":
            try:
                barrier.wait(timeout=30)
            except Exception:
                out_q.put((idx, claimed, fails, rename_errors, "barrier"))
                return
        src = os.path.join(src_dir, name)
        dst = os.path.join(dst_dir, "%s.w%d" % (name, idx))
        try:
            os.rename(src, dst)
        except OSError:
            rename_errors += 1
            continue
        if mode != "plain":
            ok, why = try_open(dst, retry=(mode == "retry"))
            if not ok:
                fails[why] = fails.get(why, 0) + 1
                continue
        claimed.append(name)
    out_q.put((idx, claimed, fails, rename_errors, "ok"))


def one_round(mode, sync, procs, nfiles, base, deadline):
    src_dir = os.path.join(base, "src")
    dst_dir = os.path.join(base, "dst")
    os.makedirs(src_dir)
    os.makedirs(dst_dir)
    names = ["task_%03d.txt" % i for i in range(nfiles)]
    for n in names:
        with open(os.path.join(src_dir, n), "wb") as f:
            f.write(b"x")
    ctx = mp.get_context("spawn")
    barrier = ctx.Barrier(procs)
    out_q = ctx.Queue()
    ps = [ctx.Process(target=worker,
                      args=(i, mode, sync, src_dir, dst_dir, names, barrier, out_q))
          for i in range(procs)]
    for p in ps:
        p.start()
    results = []
    status = "ok"
    while len(results) < procs:
        if time.time() > deadline:
            status = "deadline"
            break
        try:
            results.append(out_q.get(timeout=1.0))
        except Exception:
            pass
    for p in ps:
        p.join(timeout=5)
        if p.is_alive():
            p.terminate()  # our own child, started above
            p.join(timeout=5)
    counts = {}
    fails = {}
    for _idx, claimed, f, _re, st in results:
        if st != "ok":
            status = st
        for k, v in f.items():
            fails[k] = fails.get(k, 0) + v
        for n in claimed:
            counts[n] = counts.get(n, 0) + 1
    present = set(os.listdir(dst_dir))
    on_disk = set(p.rsplit(".w", 1)[0] for p in present)
    once = sum(1 for c in counts.values() if c == 1)
    twice = sum(1 for c in counts.values() if c >= 2)
    nobody = sum(1 for n in names if n in on_disk and n not in counts)
    return {"once": once, "twice": twice, "nobody": nobody,
            "fails": fails, "status": status}


def demo():
    print("demo (Windows only):")
    d = tempfile.mkdtemp(prefix="renamedemo_")
    try:
        p = os.path.join(d, "a.txt")
        with open(p, "wb") as f:
            f.write(b"x")
        with open(p, "rb"):
            try:
                os.rename(p, os.path.join(d, "b.txt"))
                print("  1. rename while another handle holds an ordinary "
                      "open(): succeeded")
            except OSError as e:
                print("  1. rename while another handle holds an ordinary "
                      "open(): %s, WinError %s" % (type(e).__name__,
                                                   getattr(e, "winerror", None)))
        if os.name != "nt":
            print("  2. skipped (not Windows)")
            return
        import ctypes
        from ctypes import wintypes
        k32 = ctypes.WinDLL("kernel32", use_last_error=True)
        k32.CreateFileW.restype = wintypes.HANDLE
        k32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD,
                                    wintypes.DWORD, wintypes.LPVOID,
                                    wintypes.DWORD, wintypes.DWORD,
                                    wintypes.HANDLE]
        k32.GetFinalPathNameByHandleW.argtypes = [wintypes.HANDLE,
                                                  wintypes.LPWSTR,
                                                  wintypes.DWORD,
                                                  wintypes.DWORD]
        k32.CloseHandle.argtypes = [wintypes.HANDLE]
        a = os.path.join(d, "c.txt")
        with open(a, "wb") as f:
            f.write(b"x")
        # read access; share read, write AND delete; open existing
        h = k32.CreateFileW(a, 0x80000000, 7, None, 3, 0, None)
        if h in (None, wintypes.HANDLE(-1).value):
            print("  2. could not open a delete-sharing handle")
            return
        try:
            b = os.path.join(d, "e.txt")
            os.rename(a, b)
            buf = ctypes.create_unicode_buffer(1024)
            k32.GetFinalPathNameByHandleW(h, buf, 1024, 0)
            print("  2. a handle opened before the rename with delete sharing: "
                  "rename succeeded, and the handle now reports the new name: %s"
                  % buf.value.split("\\")[-1])
        finally:
            k32.CloseHandle(h)
    finally:
        shutil.rmtree(d, ignore_errors=True)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--procs", type=int, default=5)
    ap.add_argument("--files", type=int, default=60)
    ap.add_argument("--rounds", type=int, default=10)
    ap.add_argument("--max-seconds", type=float, default=110.0)
    ap.add_argument("--no-demo", action="store_true")
    a = ap.parse_args()
    print("Python", sys.version.split()[0], "|", platform.platform(), "|",
          "cpus", os.cpu_count())
    print("procs=%d files=%d rounds-per-cell=%d" % (a.procs, a.files, a.rounds))
    if not a.no_demo:
        demo()
    root = tempfile.mkdtemp(prefix="renamerace_")
    deadline = time.time() + a.max_seconds
    cells = [(s, m) for s in ("barrier", "free")
             for m in ("plain", "verify", "retry")]
    tally = {c: [] for c in cells}
    try:
        for r in range(a.rounds):
            for sync, mode in cells:
                if time.time() > deadline:
                    continue
                base = os.path.join(root, "%s_%s_%d" % (sync, mode, r))
                tally[(sync, mode)].append(
                    one_round(mode, sync, a.procs, a.files, base, deadline))
                shutil.rmtree(base, ignore_errors=True)
    finally:
        shutil.rmtree(root, ignore_errors=True)
    print()
    print("%-8s %-7s %6s | %6s %6s %6s | failed verify-opens: "
          "not-found denied other | rounds with a double claim"
          % ("timing", "rule", "rounds", "once", "twice", "nobody"))
    other_codes = {}
    for sync, mode in cells:
        rows = tally[(sync, mode)]
        nf = sum(t["fails"].get("notfound", 0) for t in rows)
        sh = sum(t["fails"].get("denied", 0) for t in rows)
        ot = sum(v for t in rows for k, v in t["fails"].items()
                 if k.startswith("other"))
        codes = {}
        for t in rows:
            for k, v in t["fails"].items():
                if k.startswith("other"):
                    codes[k] = codes.get(k, 0) + v
        other_codes[(sync, mode)] = codes
        print("%-8s %-7s %6d | %6d %6d %6d | %24d %7d %5d | %d"
              % (sync, mode, len(rows),
                 sum(t["once"] for t in rows), sum(t["twice"] for t in rows),
                 sum(t["nobody"] for t in rows), nf, sh, ot,
                 sum(1 for t in rows if t["twice"] > 0)))
    for c, codes in other_codes.items():
        if codes:
            print("other failed opens, %s %s: %s" % (c[0], c[1], codes))
    odd = sum(1 for rows in tally.values() for t in rows if t["status"] != "ok")
    print("rounds that did not end cleanly:", odd)


if __name__ == "__main__":
    mp.freeze_support()
    main()
# Licence: MIT (see LICENSE.txt).