filelock: On Windows an oplock on a soft lock's marker holds SoftFileLock.acquire() past its timeout and blocks the stale break

"""
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:\\venvs\\rev4F\\Scripts\\python.exe oplock_vm.py fltester '<password>' C:\\tmp\\rev4F-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")

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

Run on the VM: C:\\venvs\\rev4F\\Scripts\\python.exe oplock_semantics_vm.py C:\\tmp\\rev4F-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\nhost\n")
        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'}")