"""Repro for PR #4228: silent cache corruption when flock() is a no-op (Lustre/GPFS/NFS).

Simulates the broken-filesystem behavior locally by patching `fcntl.flock` to a silent
no-op (exactly what Lustre/GPFS do: return success for every caller without blocking).

Scenario reproduced (the "silent poisoning" variant seen with multi-process vllm serve):
  - Process A starts downloading a 125MB file to `<etag>.incomplete` (opened in append mode).
  - Mid-download, process B enters the same code path. The no-op lock lets it in. B opens the
    same `.incomplete` file in append mode, sees resume_size=current size, and requests the
    missing tail via a Range request. B is a bit slower than A (here: simulated with a delay
    before its first write — in real life: network jitter, slower node, GIL pause, ...).
  - A finishes first. Its consistency check passes (B hasn't written yet), so A renames
    `.incomplete` -> blob and creates the pointer symlink. The cache looks healthy.
  - B still holds an open fd on the *renamed* inode (rename doesn't invalidate open fds) and
    appends its tail bytes to the now-live blob. The blob silently becomes larger than
    expected with duplicated bytes, while a valid pointer exists -> every subsequent reader
    gets a corrupted file. B itself fails its consistency check, but the damage is done.

Run: HF_HUB_DISABLE_XET=1 .venv/bin/python repro_4228.py
"""

import hashlib
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path

REPO_ID = "openai-community/gpt2"
FILENAME = "64-8bits.tflite"
EXPECTED_SIZE = 125162496
EXPECTED_SHA256 = "c966da3b74697803352ca7c6f2f220e7090a557b619de9da0c6b34d89f7825c1"

WORKER_A = """
import sys
import fcntl
fcntl.flock = lambda *a, **k: None  # Lustre/GPFS behavior: silent success for every caller
from huggingface_hub import hf_hub_download
path = hf_hub_download(repo_id=sys.argv[1], filename=sys.argv[2])
print("worker A done:", path)
"""

# Worker B: same no-op flock, but (1) pre-warms imports + TLS connection, then waits for a
# signal file so it can enter the download loop while A is mid-flight, and (2) delays its
# first chunk write by a few seconds to emulate a slower sibling process.
WORKER_B = """
import os
import sys
import time
import fcntl
fcntl.flock = lambda *a, **k: None  # Lustre/GPFS behavior: silent success for every caller

import huggingface_hub.file_download as fd
from huggingface_hub import hf_hub_download, get_hf_file_metadata, hf_hub_url

repo_id, filename, signal_file = sys.argv[1], sys.argv[2], sys.argv[3]

# Pre-warm DNS/TLS + imports so the post-signal latency is minimal.
get_hf_file_metadata(hf_hub_url(repo_id=repo_id, filename=filename))

_orig_http_get = fd.http_get

class SlowFirstWrite:
    def __init__(self, f):
        self._f = f
        self._first = True
    def write(self, data):
        if self._first:
            time.sleep(6)  # let A finish, pass its consistency check, and rename the blob
            self._first = False
        return self._f.write(data)
    def __getattr__(self, name):
        return getattr(self._f, name)

def slow_http_get(url, temp_file, **kwargs):
    return _orig_http_get(url, SlowFirstWrite(temp_file), **kwargs)

fd.http_get = slow_http_get

while not os.path.exists(signal_file):
    time.sleep(0.01)

path = hf_hub_download(repo_id=repo_id, filename=filename)
print("worker B done:", path)
"""


def main() -> None:
    tmp = tempfile.mkdtemp(prefix="hf_repro_4228_")
    env = {**os.environ, "HF_HOME": tmp, "HF_HUB_DISABLE_XET": "1"}
    signal_file = os.path.join(tmp, "go_b")
    print(f"HF_HOME={tmp}")

    repo_dir = Path(tmp) / "hub" / "models--openai-community--gpt2"

    b = subprocess.Popen(
        [sys.executable, "-c", WORKER_B, REPO_ID, FILENAME, signal_file],
        env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
    )
    time.sleep(3)  # let B pre-warm
    a = subprocess.Popen(
        [sys.executable, "-c", WORKER_A, REPO_ID, FILENAME],
        env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
    )

    # Signal B once A's download is ~half way.
    deadline = time.time() + 300
    while time.time() < deadline:
        candidates = list(repo_dir.glob("blobs/*.incomplete")) if repo_dir.exists() else []
        if candidates and candidates[0].stat().st_size > 0.85 * EXPECTED_SIZE:
            print(f"incomplete at {candidates[0].stat().st_size / EXPECTED_SIZE:.0%} -> signaling worker B")
            Path(signal_file).touch()
            break
        if a.poll() is not None:
            print("worker A finished too early; rerun")
            print(a.stdout.read())
            b.kill()
            return
        time.sleep(0.02)

    out_a, _ = a.communicate()
    print(f"--- worker A (exit {a.returncode}) ---\n{out_a}")

    blob = repo_dir / "blobs" / EXPECTED_SHA256
    if blob.exists():
        print(f"blob right after A finished: size={blob.stat().st_size} (expected {EXPECTED_SIZE}) -> looks healthy")

    out_b, _ = b.communicate()
    print(f"--- worker B (exit {b.returncode}) ---\n{out_b}")

    pointer = next((repo_dir / "snapshots").rglob(FILENAME), None) if (repo_dir / "snapshots").exists() else None
    print(f"pointer: {pointer}")
    if blob.exists():
        actual_size = blob.stat().st_size
        actual_sha = hashlib.sha256(blob.read_bytes()).hexdigest()
        print(f"final blob size: {actual_size} (expected {EXPECTED_SIZE})")
        print(f"final blob sha256: {actual_sha}\n     expected sha256: {EXPECTED_SHA256}")
        if pointer is not None and (actual_size != EXPECTED_SIZE or actual_sha != EXPECTED_SHA256):
            print("\n*** SILENT CACHE POISONING: corrupted blob behind a valid pointer ***")
        elif actual_size == EXPECTED_SIZE and actual_sha == EXPECTED_SHA256:
            print("\nblob is intact (race not triggered this run, try again)")
    else:
        print(f"blob missing: {blob}")


if __name__ == "__main__":
    main()