"""
Windows oplock checks beyond the lock-file open PR #835 changes: release() and the soft locks' marker read.

Run on the VM (PowerShell, from the synced worktree, with the fltester account present):
 C:venvsrev4FScriptspython.exe oplock vm.py fltester '<password>' C:tmprev4F-oplock

[1] A second account keeps requesting a Level 1 oplock on the lock file through a read handle that shares delete.
  WindowsFileLock. release closes its handle and then unlinks by path; an oplock granted in between makes the
  DeleteFileW inside os.unlink wait for the holder's acknowledgement. [1a] races the close-to-unlink window as is;
  [1b] widens that window to 200 ms with a patched os.unlink so the grant lands inside it.
[2] A second account holds a Level 1 oplock on a soft-lock marker whose holder is alive. SoftFileLock.acquire reads the
  marker through os.open, which waits for the acknowledgement, past timeout=1.

A thread that exits has its pending oplock request cancelled, so every holder thread stays alive until the end.
"""

from  future  import annotations

import ctypes
import os
import subprocess
import sys
import threading
import time
from ctypes import wintypes

from filelock import FileLock, SoftFileLock, Timeout

user, password, base = sys.argv[1], sys.argv[2], sys.argv[3]
icacls = os.path.join(os.environ["SYSTEMROOT"], "System32", "icacls.exe")
os.makedirs(base, exist ok=True)
subprocess.run([icacls, base, "/grant", f"{user}:(OI)(CI)(M)"], check=True, capture output=True)

advapi32 = ctypes.WinDLL("advapi32", use last error=True)
kernel32 = ctypes.WinDLL("kernel32", use last error=True)
advapi32.LogonUserW.argtypes = [
  wintypes.LPCWSTR,
  wintypes.LPCWSTR,
  wintypes.LPCWSTR,
  wintypes.DWORD,
  wintypes.DWORD,
  ctypes.POINTER(wintypes.HANDLE),
]
kernel32.CreateFileW.restype = kernel32.CreateEventW.restype = wintypes.HANDLE
kernel32.CreateFileW.argtypes = [
  wintypes.LPCWSTR,
  wintypes.DWORD,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  wintypes.DWORD,
  wintypes.HANDLE,
]


class Overlapped(ctypes.Structure):
   fields = (
    ("Internal", ctypes.c void p),
    ("InternalHigh", ctypes.c void p),
    ("Offset", wintypes.DWORD),
    ("OffsetHigh", wintypes.DWORD),
    ("hEvent", wintypes.HANDLE),
  )


kernel32.DeviceIoControl.argtypes = [
  wintypes.HANDLE,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  ctypes.c void p,
  ctypes.POINTER(Overlapped),
]
kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]

GENERIC READ = 0x80000000
SHARE ALL = 0x7
OPEN EXISTING = 3
FLAG OVERLAPPED = 0x40000000
INVALID HANDLE = wintypes.HANDLE(-1).value
FSCTL REQUEST OPLOCK LEVEL 1 = 0x00090000
ERROR IO PENDING = 997
hold = threading.Event()
real unlink = os.unlink


def impersonate() -> None:
  token = wintypes.HANDLE()
  if not advapi32.LogonUserW(user, ".", password, 3, 0, ctypes.byref(token)):
    raise ctypes.WinError(ctypes.get last error())
  if not advapi32.ImpersonateLoggedOnUser(token):
    raise ctypes.WinError(ctypes.get last error())


def request level1(path: str) -> tuple[int, Overlapped] | None:
  handle = kernel32.CreateFileW(path, GENERIC READ, SHARE ALL, None, OPEN EXISTING, FLAG OVERLAPPED, None)
  if handle == INVALID HANDLE:
    return None
  overlapped = Overlapped(hEvent=kernel32.CreateEventW(None, True, False, None))
  if (
    not kernel32.DeviceIoControl(handle, FSCTL REQUEST OPLOCK LEVEL 1, None, 0, None, 0, None, ctypes.byref(overlapped))
    and ctypes.get last error() == ERROR IO PENDING
  ):
    return handle, overlapped
  kernel32.CloseHandle(handle)
  kernel32.CloseHandle(overlapped.hEvent)
  return None


def drop(granted: tuple[int, Overlapped]) -> None:
  handle, overlapped = granted
  kernel32.CloseHandle(handle)
  kernel32.WaitForSingleObject(overlapped.hEvent, 5000)
  kernel32.CloseHandle(overlapped.hEvent)


def release under oplock requests(label: str, lock path: str, cycles: int) -> bool:
  stop = threading.Event()
  granted box: list[tuple[int, Overlapped]] = []

  def spinner() -> None:
    impersonate()
    while not stop.is set():
      if (granted := request level1(lock path)) is not None:
        granted box.append(granted)
        hold.wait()
        return

  releasing = threading.Event()
  release durations: list[float] = []

  def owner() -> None:
    # thread local locks release only on the acquiring thread, so acquire and release stay on this one
    for  in range(cycles):
      lock = FileLock(lock path, is singleton=False)
      lock.acquire(timeout=5)
      started = time.monotonic()
      releasing.set()
      lock.release()
      releasing.clear()
      release durations.append(time.monotonic() - started)
      if granted box:
        return

  holder = threading.Thread(target=spinner, daemon=True)
  holder.start()
  owner thread = threading.Thread(target=owner, daemon=True)
  owner thread.start()
  owner thread.join(60)
  stalled = owner thread.is alive() and releasing.is set()
  stop.set()
  longest = max(release durations, default=0.0) * 1000
  print(f" {label}: oplock granted: {bool(granted box)}; cycles run: {len(release durations)}; longest release: {longest:.1f} ms")
  if stalled:
    print(" release() still running after 60 s while the oplock holder has not answered the break")
    if granted box:
      drop(granted box[0])
    owner thread.join(10)
    print(f" release() returned after the holder closed its handle: {not owner thread.is alive()}")
  return stalled


print("[1a] release() against a second account that keeps requesting a Level 1 oplock, natural window")
stalled native = release under oplock requests("natural", os.path.join(base, "native.lock"), 500)

print("[1b] the same, with the window between the handle close and the unlink widened to 200 ms")


def slow unlink(path: str, *, dir fd: int | None = None) -> None:
  time.sleep(0.2)
  real unlink(path, dir fd=dir fd)


os.unlink = slow unlink
stalled scheduled = release under oplock requests("scheduled", os.path.join(base, "native-scheduled.lock"), 20)
os.unlink = real unlink

print("[2] SoftFileLock.acquire(timeout=1) against a second account's Level 1 oplock on a live marker")
marker = os.path.join(base, "soft.lock")
holder = SoftFileLock(marker, is singleton=False)
holder.acquire()
holder fd = holder. context.lock file fd # noqa: SLF001
os.close(holder fd) # the marker stays with a live PID, as after a holder that lost only its handle
impersonate done = threading.Event()
granted soft: list[tuple[int, Overlapped]] = []


def take soft oplock() -> None:
  impersonate()
  if (granted := request level1(marker)) is not None:
    granted soft.append(granted)
  impersonate done.set()
  hold.wait()


threading.Thread(target=take soft oplock, daemon=True).start()
impersonate done.wait(5)
print(f" oplock granted: {bool(granted soft)}")
outcome = ["still running after 25 s"]


def contend() -> None:
  started = time.monotonic()
  try:
    SoftFileLock(marker, is singleton=False).acquire(timeout=1)
    outcome[0] = f"acquired after {time.monotonic() - started:.2f} s"
  except Timeout:
    outcome[0] = f"Timeout after {time.monotonic() - started:.2f} s"
  except OSError as error:
    outcome[0] = f"{type(error). name } after {time.monotonic() - started:.2f} s: {error}"


contender = threading.Thread(target=contend, daemon=True)
contender.start()
contender.join(25)
print(f" contender: {outcome[0]}")
soft stalled = "still running" in outcome[0]
if granted soft:
  drop(granted soft[0])
  contender.join(10)
  print(f" after the holder closed: {outcome[0]}")
hold.set()
print("STALLS" if stalled native or stalled scheduled or soft stalled else "BOUNDED")

[2] SoftFileLock.acquire(timeout=1) against a second account's Level 1 oplock on a live marker
 oplock granted: True
 contender: Timeout after 10.02 s
 after the holder closed: Timeout after 10.02 s

"""
Which path operations wait for an unanswered oplock break on Windows 11 NTFS. Same account, one process.

Run on the VM: C:venvsrev4FScriptspython.exe oplock semantics vm.py C:tmprev4F-sem
"""

from  future  import annotations

import ctypes
import os
import sys
import threading
import time
from ctypes import wintypes

base = sys.argv[1]
os.makedirs(base, exist ok=True)
kernel32 = ctypes.WinDLL("kernel32", use last error=True)
kernel32.CreateFileW.restype = kernel32.CreateEventW.restype = wintypes.HANDLE
kernel32.CreateFileW.argtypes = [
  wintypes.LPCWSTR,
  wintypes.DWORD,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  wintypes.DWORD,
  wintypes.HANDLE,
]


class Overlapped(ctypes.Structure):
   fields = (
    ("Internal", ctypes.c void p),
    ("InternalHigh", ctypes.c void p),
    ("Offset", wintypes.DWORD),
    ("OffsetHigh", wintypes.DWORD),
    ("hEvent", wintypes.HANDLE),
  )


kernel32.DeviceIoControl.argtypes = [
  wintypes.HANDLE,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  ctypes.c void p,
  wintypes.DWORD,
  ctypes.c void p,
  ctypes.POINTER(Overlapped),
]
kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]
GENERIC READ, GENERIC WRITE = 0x80000000, 0x40000000
SHARE ALL, OPEN EXISTING, FLAG OVERLAPPED = 0x7, 3, 0x40000000
INVALID HANDLE = wintypes.HANDLE(-1).value
LEVELS = {"level1": 0x00090000, "batch": 0x00090008}


def take(path: str, level: str) -> tuple[int, Overlapped]:
  handle = kernel32.CreateFileW(path, GENERIC READ, SHARE ALL, None, OPEN EXISTING, FLAG OVERLAPPED, None)
  overlapped = Overlapped(hEvent=kernel32.CreateEventW(None, True, False, None))
  pending = not kernel32.DeviceIoControl(handle, LEVELS[level], None, 0, None, 0, None, ctypes.byref(overlapped))
  assert pending and ctypes.get last error() == 997, ctypes.get last error()
  return handle, overlapped


def drop(granted: tuple[int, Overlapped]) -> None:
  kernel32.CloseHandle(granted[0])
  kernel32.WaitForSingleObject(granted[1].hEvent, 5000)
  kernel32.CloseHandle(granted[1].hEvent)


def timed(label: str, operation) -> None: # noqa: ANN001
  done = threading.Event()
  result = ["still running after 15 s"]

  def run() -> None:
    started = time.monotonic()
    try:
      operation()
      result[0] = f"returned after {(time.monotonic() - started) * 1000:.1f} ms"
    except OSError as error:
      result[0] = f"{type(error). name } after {(time.monotonic() - started) * 1000:.1f} ms: {error}"
    done.set()

  threading.Thread(target=run, daemon=True).start()
  done.wait(15)
  print(f" {label}: {result[0]}")
  return done


def open read(path: str) -> None:
  os.close(os.open(path, os.O RDONLY | os.O BINARY))


def open write(path: str) -> None:
  os.close(os.open(path, os.O RDWR | os.O BINARY))


def create new(path: str) -> None:
  os.close(os.open(path, os.O WRONLY | os.O CREAT | os.O EXCL | os.O BINARY))


for level in LEVELS:
  for label, operation, consumes in (
    ("os.open O RDONLY", open read, False),
    ("os.open O RDWR", open write, False),
    ("os.open O CREAT|O EXCL on the existing file", create new, False),
    ("os.utime by path", lambda p: os.utime(p, None), False),
    ("os.rename", lambda p: os.rename(p, p + ".moved"), True),
    ("os.unlink", os.unlink, True),
  ):
    path = os.path.join(base, f"{level}-{label.replace(' ', ' ').replace('|', '')}.lock")
    with open(path, "w", encoding="utf-8") as handle:
      handle.write("1
host
")
    granted = take(path, level)
    print(f"{level}, {label}")
    done = timed(label, lambda p=path, op=operation: op(p))
    drop(granted)
    done.wait(5)
    print(f"  after the holder closed: {'done' if done.is set() else 'still running'}")

sequenceDiagram
  participant A as other local account
  participant M as soft.lock marker, no handle open
  participant L as lock user: SoftFileLock(path).acquire(timeout=1)
  participant W as Windows (NTFS)
  A->>W: CreateFileW(soft.lock, GENERIC READ, share read/write/delete, FILE FLAG OVERLAPPED)
  A->>W: DeviceIoControl(FSCTL REQUEST OPLOCK LEVEL 1), granted, break never acknowledged
  L->>W: os.open(soft.lock, O CREAT|O EXCL) fails with EEXIST at once
  L->>W: os.lstat(soft.lock) regular file
  rect rgb(255, 214, 214)
    L->>W: os.open(soft.lock, O RDONLY) inside read lock file
    Note over L,W: CreateFileW asks A to break the oplock and waits#59; the 1 s deadline, blocking=False and cancel check are not consulted
    W-->>L: handle after 10.02 s (two-account run), or when A closes (same-account probe)
  end
  L->>L: check give up: past the deadline, raise Timeout after 10.02 s
  Note over A,M: a stale marker takes the same wait in Path.rename and Path.unlink, so it is not broken