Function scan_completion_lock

plugins/codex-security/scripts/workbench_db.py:326–345

(scan_id: str)

@contextmanager
def scan_completion_lock(scan_id: str) -> Any:
    lock_dir = state_dir() / "completion-locks"
    lock_dir.mkdir(parents=True, exist_ok=True)
    lock_path = lock_dir / f"{require_uuid(scan_id, 'scan-id')}.lock"
    descriptor = os.open(
        lock_path,
        os.O_RDWR | os.O_CREAT | getattr(os, "O_BINARY", 0),
        0o600,
    )
    locked = False
    try:
        acquire_completion_file_lock(descriptor)
        locked = True
        yield
    finally:
        try:
            if locked:
                release_completion_file_lock(descriptor)
        finally:
            os.close(descriptor)


def is_file_lock_contention(error: OSError) -> bool: