mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-10-23 06:30:28 +08:00
Mesut Oezdil
f084eb39cb
* fix(sbom): release lock before sleeping in _rate_limit time.sleep was called inside the _rate_lock block, blocking all threads from checking their own domain rate limit while one thread slept. With _MAX_WORKERS=12 querying crates.io, npm, and pypi concurrently, this made the thread pool effectively serial. Move the sleep outside the lock so threads for different domains can proceed concurrently. * fix(sbom): reserve next slot in _rate_limit to prevent same-domain races Use time.monotonic() and store next_req = max(now, last + interval) so concurrent same-domain callers each get a unique future slot rather than all sleeping the same amount. Add regression tests for same-domain spacing and different-domain non-blocking. * fix(sbom): wire rate-limit tests into test suite via test:sbom task Rename test_resolve_licenses.py to resolve_licenses_test.py to match repo *_test.py convention. Add test:sbom task to tasks/test.toml and include it in the top-level test depends so mise run test picks it up. * fix(sbom): add SPDX header, fix cross-domain regression test to exercise lock contention * fix(sbom): replace zip with itertools.pairwise to satisfy ruff lint
69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from resolve_licenses import _last_request, _rate_limit, _rate_lock
|
|
|
|
|
|
def test_same_domain_requests_are_spaced() -> None:
|
|
domain = "test.same-domain.example"
|
|
with _rate_lock:
|
|
_last_request.pop(domain, None)
|
|
|
|
interval = 0.05
|
|
times: list[float] = []
|
|
|
|
def call() -> None:
|
|
_rate_limit(domain, interval=interval)
|
|
times.append(time.monotonic())
|
|
|
|
with ThreadPoolExecutor(max_workers=3) as pool:
|
|
list(pool.map(lambda _: call(), range(3)))
|
|
|
|
times.sort()
|
|
for a, b in itertools.pairwise(times):
|
|
assert b - a >= interval * 0.9, f"gap {b - a:.4f}s < interval {interval}s"
|
|
|
|
|
|
def test_different_domains_do_not_block_each_other() -> None:
|
|
alpha = "alpha2.example"
|
|
beta = "beta2.example"
|
|
interval = 0.1
|
|
|
|
now = time.monotonic()
|
|
with _rate_lock:
|
|
_last_request[alpha] = now # alpha must sleep for ~interval
|
|
_last_request.pop(beta, None) # beta is free
|
|
|
|
ready = threading.Event()
|
|
|
|
def call_alpha() -> None:
|
|
ready.set()
|
|
_rate_limit(alpha, interval=interval)
|
|
|
|
t = threading.Thread(target=call_alpha)
|
|
t.start()
|
|
ready.wait()
|
|
time.sleep(0.01) # let alpha enter its sleep
|
|
|
|
beta_start = time.monotonic()
|
|
_rate_limit(beta, interval=interval)
|
|
beta_elapsed = time.monotonic() - beta_start
|
|
|
|
t.join()
|
|
|
|
assert beta_elapsed < interval * 0.5, f"beta blocked for {beta_elapsed:.3f}s"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_same_domain_requests_are_spaced()
|
|
print("same-domain spacing: ok")
|
|
test_different_domains_do_not_block_each_other()
|
|
print("different-domain non-blocking: ok")
|