Reduce history requests and protect hidden local edits

This commit is contained in:
AngusBell97
2026-09-11 14:57:14 +01:00
committed by firestar5683
parent c4f46c51b2
commit f88ef758ca
5 changed files with 493 additions and 35 deletions
@@ -313,17 +313,22 @@ def test_invalid_version_metadata_is_retryable(history, repo, api, monkeypatch,
URLError('offline'), IncompleteRead(b'partial', 100),
])
def test_version_network_failure_is_not_cached_as_missing(history, repo, api, monkeypatch, failure):
clock = [1000.0]
monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
monkeypatch.setattr(history.time, 'time', lambda: clock[0])
api.heads['StarPilot'] = 1
def fail(*args, **kwargs):
raise failure
monkeypatch.setattr(history, '_open_raw_url', fail, raising=False)
with pytest.raises(history.VersionHistoryError, match='retry'):
history.list_versions(repo, 'StarPilot')
clock[0] += 61
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
assert history.list_versions(repo, 'StarPilot')['commits'][0]['version'] == '6.7.7'
def test_version_cache_outlives_branch_cache_and_expires(history, repo, api, monkeypatch):
monkeypatch.setattr(history, '_saved_display_versions', lambda *a: {}) # Exercise memory-cache expiry alone.
api.heads['StarPilot'] = 1
clock = [100.0]
monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
@@ -488,7 +493,7 @@ def test_rate_limit_reports_retry_time(history, monkeypatch, headers, expected):
@pytest.mark.parametrize('code', [403, 429])
def test_rate_limit_short_backoff_is_shared_and_then_retries(history, monkeypatch, code):
def test_rate_limit_full_backoff_is_shared_and_then_retries(history, monkeypatch, code):
clock = [1000.0]
monkeypatch.setattr(history.time, 'time', lambda: clock[0])
monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
@@ -504,6 +509,10 @@ def test_rate_limit_short_backoff_is_shared_and_then_retries(history, monkeypatc
history._get_json('https://api.github.com/repos/a/b/' + path)
assert len(calls) == 1
clock[0] += 61
with pytest.raises(history.HistoryUnavailable):
history._get_json('https://api.github.com/repos/a/b/branches/main')
assert len(calls) == 1
clock[0] = 3280
assert history._get_json('https://api.github.com/repos/a/b/branches/main') == {'ok': True}
assert len(calls) == 2
@@ -713,6 +722,7 @@ def test_starpilot_snapshot_fallback_only_for_version_transport_failure(history,
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
first = history.list_versions(repo, 'StarPilot')
offline = offline_reload(history, monkeypatch)
monkeypatch.setattr(offline, '_saved_display_versions', lambda *a: {}) # Force transport to test failure policy.
monkeypatch.setattr(offline, '_get_json', api)
def raw(*args, **kwargs):
if failure_kind == 'offline':
@@ -773,6 +783,7 @@ def test_http_failure_snapshot_fallback_boundary(history, repo, api, monkeypatch
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
first = history.list_versions(repo, branch)
offline = offline_reload(history, monkeypatch)
monkeypatch.setattr(offline, '_saved_display_versions', lambda *a: {}) # Force transport to test failure policy.
def fail(*args, **kwargs):
raise HTTPError('https://github.com/', code, 'Unavailable', headers, io.BytesIO())
if transport == 'api':
@@ -792,3 +803,58 @@ def test_http_failure_snapshot_fallback_boundary(history, repo, api, monkeypatch
def test_empty_snapshot_preload_timestamp_is_rejected(history, repo, api):
first = history.list_versions(repo, 'main')
assert not history._save_history_snapshot('https://api.github.com/repos/firestar5683/openpilot', 'main', 1, None, first, saved_at='')
def test_saved_release_labels_survive_restart_without_repeating_raw_downloads(history, repo, api, monkeypatch):
api.heads['StarPilot'] = 3
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
first = history.list_versions(repo, 'StarPilot')
restarted = offline_reload(history, monkeypatch)
monkeypatch.setattr(restarted, '_get_json', api)
def forbidden(*args, **kwargs):
pytest.fail('A saved immutable release label was downloaded again')
monkeypatch.setattr(restarted, '_open_raw_url', forbidden)
assert restarted.list_versions(repo, 'StarPilot')['commits'] == first['commits']
# A changed head cannot reuse the previous first-page snapshot blindly.
api.heads['StarPilot'] = 4
restarted._cache.clear()
calls = []
def raw(*args, **kwargs):
calls.append(1)
return io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.8"\n')
monkeypatch.setattr(restarted, '_open_raw_url', raw)
result = restarted.list_versions(repo, 'StarPilot')
assert result['head'] == f'{4:040x}' and result['commits'][0]['version'] == '6.7.8'
assert calls
def test_saved_labels_never_bypass_live_history_validation(history, repo, api, monkeypatch):
api.heads['StarPilot'] = 3
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n'))
first = history.list_versions(repo, 'StarPilot')
restarted = offline_reload(history, monkeypatch)
def invalid(url):
if '/commits?' in url: return {'invalid': True}
return api(url)
monkeypatch.setattr(restarted, '_get_json', invalid)
with pytest.raises(restarted.VersionHistoryError, match='invalid commit history'):
restarted.list_versions(repo, 'StarPilot')
def test_nine_page_release_browse_reuses_saved_labels_after_restart(history, repo, api, monkeypatch):
api.heads['StarPilot'] = 240
calls = []
def raw(*args, **kwargs):
calls.append(1)
return io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"\n')
monkeypatch.setattr(history, '_open_raw_url', raw)
head = f'{240:040x}'
for page in range(1, 10):
history.list_versions(repo, 'StarPilot', page=page, head=head if page > 1 else None)
assert len(calls) == 225
assert len(api.requests) == 4 # One branch head and three 100-commit blocks.
restarted = offline_reload(history, monkeypatch)
monkeypatch.setattr(restarted, '_get_json', api)
monkeypatch.setattr(restarted, '_open_raw_url', raw)
for page in range(1, 10):
restarted.list_versions(repo, 'StarPilot', page=page, head=head if page > 1 else None)
assert len(calls) == 225 # No repeat raw-file requests for those 225 saved labels.
assert len(api.requests) == 8 # The live branch and history are still validated.
@@ -0,0 +1,109 @@
"""Refuse destructive checkout when ordinary patches cannot preserve local files."""
import pytest
from test_version_install_index import checkout, git, install, installer, repository_snapshot
@pytest.mark.parametrize('flag', ['assume-unchanged', 'skip-worktree'])
def test_hidden_index_flags_refuse_install_without_mutation(checkout, flag):
repo, data, old, latest = checkout
git(repo, 'update-index', '--' + flag, 'tracked.txt')
(repo / 'tracked.txt').write_bytes(b'valuable hidden working edit\n')
before = repository_snapshot(repo, data)
flags = git(repo, 'ls-files', '-v', '-z')
with pytest.raises(installer.InstallError, match='assume-unchanged|skip-worktree'):
install(checkout)
assert repository_snapshot(repo, data) == before
assert git(repo, 'ls-files', '-v', '-z') == flags
assert not (data / 'starpilot/version-backups').exists()
@pytest.mark.parametrize('flag', ['assume-unchanged', 'skip-worktree'])
def test_hidden_index_flags_refuse_restore_without_mutation(checkout, flag):
repo, data, old, latest = checkout
backup = install(checkout)
git(repo, 'update-index', '--' + flag, 'tracked.txt')
(repo / 'tracked.txt').write_bytes(b'valuable hidden post-install edit\n')
before = repository_snapshot(repo, data)
flags = git(repo, 'ls-files', '-v', '-z')
with pytest.raises(installer.InstallError, match='assume-unchanged|skip-worktree'):
installer.restore(backup, check_parked=lambda: installer.require_parked(data))
assert repository_snapshot(repo, data) == before
assert git(repo, 'ls-files', '-v', '-z') == flags
def ignored_checkout(checkout, shape):
repo, data, old, latest = checkout
target_path = 'hidden-local/child.txt' if shape == 'ancestor' else 'hidden-local'
target = repo / target_path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(b'target committed content\n')
git(repo, 'add', target_path)
git(repo, 'commit', '-m', 'Target containing collision path')
target_sha = git(repo, 'rev-parse', 'HEAD').decode().strip()
git(repo, 'rm', target_path)
git(repo, 'commit', '-m', 'Current without collision path')
current = git(repo, 'rev-parse', 'HEAD').decode().strip()
(repo / '.git/info/exclude').write_text('hidden-local\n')
local = repo / ('hidden-local/child.txt' if shape == 'descendant' else 'hidden-local')
local.parent.mkdir(parents=True, exist_ok=True)
outside = data / 'outside-sentinel'
outside.write_bytes(b'valuable local ignored content\n')
if shape == 'symlink':
local.symlink_to(outside)
else:
local.write_bytes(outside.read_bytes())
return (repo, data, target_sha, current), local, outside
@pytest.mark.parametrize('shape', ['exact', 'ancestor', 'descendant', 'symlink'])
def test_ignored_conflicts_refuse_install_before_backup(checkout, shape):
state, local, outside = ignored_checkout(checkout, shape)
repo, data, old, latest = state
before = repository_snapshot(repo, data)
with pytest.raises(installer.InstallError, match='[Ii]gnored'):
install(state)
assert repository_snapshot(repo, data) == before
assert local.read_bytes() == outside.read_bytes() == b'valuable local ignored content\n'
assert local.is_symlink() == (shape == 'symlink')
assert not (data / 'starpilot/version-backups').exists()
@pytest.mark.parametrize('shape', ['exact', 'ancestor', 'descendant', 'symlink'])
def test_ignored_conflicts_refuse_restore_without_mutation(checkout, shape):
state, local, outside = ignored_checkout(checkout, shape)
repo, data, target, current = state
# Save source that tracks the conflicting path, then mimic a later checkout
# with a newly created ignored file. Restore must inspect that live state.
if local.is_symlink() or local.is_file():
local.unlink()
else:
raise AssertionError('Expected a file or symlink')
if shape == 'descendant':
local.parent.rmdir()
git(repo, 'checkout', '--force', '-B', 'Dom', target)
backup = installer._backup(repo, {'branch': 'Dom', 'commit': current, 'pinned': False}, data)
git(repo, 'checkout', '--force', '-B', 'Dom', current)
local.parent.mkdir(parents=True, exist_ok=True)
if shape == 'symlink':
local.symlink_to(outside)
else:
local.write_bytes(outside.read_bytes())
before = repository_snapshot(repo, data)
with pytest.raises(installer.InstallError, match='[Ii]gnored'):
installer.restore(backup, check_parked=lambda: installer.require_parked(data))
assert repository_snapshot(repo, data) == before
assert local.read_bytes() == outside.read_bytes() == b'valuable local ignored content\n'
assert local.is_symlink() == (shape == 'symlink')
def test_nonconflicting_ignored_files_allow_install_and_restore(checkout):
repo, data, old, latest = checkout
(repo / '.git/info/exclude').write_text('local-cache/\ntracked.txt.extra\n')
paths = [repo / 'local-cache/data.bin', repo / 'tracked.txt.extra']
for path in paths:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b'kept ignored content\n')
backup = install(checkout)
installer.restore(backup, check_parked=lambda: installer.require_parked(data))
assert all(path.read_bytes() == b'kept ignored content\n' for path in paths)
@@ -0,0 +1,191 @@
"""Offline quota and request-coalescing regressions."""
import importlib.util
import io
import threading
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.error import HTTPError
import pytest
@pytest.fixture
def history():
path = Path(__file__).resolve().parents[1] / 'version_history.py'
spec = importlib.util.spec_from_file_location('version_rate_limits_under_test', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.mark.parametrize('transport', ['api', 'raw'])
@pytest.mark.parametrize('headers,deadline', [({'Retry-After': '180'}, 1180),
({'X-RateLimit-Remaining': '0', 'X-RateLimit-Reset': '4600'}, 4600), ({}, 1060)])
def test_full_server_backoff_blocks_all_urls_until_deadline(history, monkeypatch, transport, headers, deadline):
clock = [1000.0]
monkeypatch.setattr(history.time, 'time', lambda: clock[0])
monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
calls = []
def limited(request, **kwargs):
calls.append(request.full_url)
if len(calls) == 1:
raise HTTPError(request.full_url, 429, 'Limited', headers, io.BytesIO())
return io.BytesIO(b'{}' if transport == 'api' else b'STARPILOT_DISPLAY_VERSION = "6.7.7"')
monkeypatch.setattr(history, '_open_url' if transport == 'api' else '_open_raw_url', limited)
fetch = history._get_json if transport == 'api' else history._get_display_version
with pytest.raises(history.HistoryUnavailable):
fetch('https://example.test/first')
clock[0] = deadline - 1
with pytest.raises(history.HistoryUnavailable):
fetch('https://example.test/second')
assert len(calls) == 1
clock[0] = deadline
fetch('https://example.test/second')
assert len(calls) == 2
@pytest.mark.parametrize('transport', ['api', 'raw'])
def test_concurrent_identical_cache_misses_download_once(history, monkeypatch, transport):
start = threading.Barrier(3)
entered, duplicate, release = threading.Event(), threading.Event(), threading.Event()
calls = []
def response(*args, **kwargs):
calls.append(1)
entered.set()
if len(calls) > 1:
duplicate.set()
assert release.wait(timeout=0.8)
return {} if transport == 'api' else '6.7.7'
monkeypatch.setattr(history, '_get_json' if transport == 'api' else '_get_display_version', response)
def caller():
start.wait(timeout=0.8)
if transport == 'api':
return history._json('https://api.github.com/repos/a/b/commits')
return history._display_version('https://api.github.com/repos/a/b', 'a' * 40)
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(caller) for _ in range(2)]
start.wait(timeout=0.8)
try:
assert entered.wait(timeout=0.8)
duplicate.wait(timeout=0.1)
finally:
release.set()
assert futures[0].result(timeout=0.8) == futures[1].result(timeout=0.8)
assert len(calls) == 1
def test_raw_quota_does_not_block_api_or_cached_raw(history, monkeypatch):
base, sha = 'https://api.github.com/repos/a/b', 'a' * 40
monkeypatch.setattr(history, '_open_raw_url', lambda *a, **kw: io.BytesIO(b'STARPILOT_DISPLAY_VERSION = "6.7.7"'))
assert history._display_version(base, sha) == '6.7.7'
def limited(request, **kwargs):
raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
monkeypatch.setattr(history, '_open_raw_url', limited)
with pytest.raises(history.HistoryUnavailable):
history._display_version(base, 'b' * 40)
assert history._display_version(base, sha) == '6.7.7'
monkeypatch.setattr(history, '_open_url', lambda *a, **kw: io.BytesIO(b'{}'))
assert history._get_json(base) == {}
def test_fresh_api_calls_do_not_reuse_cached_head(history, monkeypatch):
calls = []
def response(url):
calls.append(url)
return {'call': len(calls)}
monkeypatch.setattr(history, '_get_json', response)
assert history._json('head') == {'call': 1}
assert history._json('head', fresh=True) == {'call': 2}
assert history._json('head', fresh=True) == {'call': 3}
def test_later_shorter_raw_response_cannot_shorten_active_backoff(history, monkeypatch):
clock = [1000.0]
monkeypatch.setattr(history.time, 'time', lambda: clock[0])
monkeypatch.setattr(history.time, 'monotonic', lambda: clock[0])
started, long_finished = threading.Barrier(2), threading.Event()
calls = []
def limited(request, **kwargs):
calls.append(request.full_url)
started.wait(timeout=0.8)
if request.full_url.endswith('short'):
assert long_finished.wait(timeout=0.8)
wait = '180' if request.full_url.endswith('long') else '60'
raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': wait}, io.BytesIO())
monkeypatch.setattr(history, '_open_raw_url', limited)
def caller(suffix):
with pytest.raises(history.HistoryUnavailable):
history._get_display_version('https://raw.githubusercontent.com/' + suffix)
if suffix == 'long':
long_finished.set()
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(caller, suffix) for suffix in ['long', 'short']]
for future in futures:
future.result(timeout=0.8)
clock[0] = 1179
with pytest.raises(history.HistoryUnavailable):
history._get_display_version('https://raw.githubusercontent.com/third')
assert len(calls) == 2
def test_raw_worker_waiting_for_slot_observes_new_quota(history, monkeypatch):
entered, queued, release = threading.Event(), threading.Event(), threading.Event()
slots, gate_lock = threading.Semaphore(1), threading.Lock()
counts = {'attempts': 0, 'http': 0}
class Gate:
def __enter__(self):
with gate_lock:
counts['attempts'] += 1
if counts['attempts'] == 2:
queued.set()
assert slots.acquire(timeout=0.8)
def __exit__(self, *args):
slots.release()
monkeypatch.setattr(history, '_version_slots', Gate())
def limited(request, **kwargs):
counts['http'] += 1
entered.set()
assert release.wait(timeout=0.8)
raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
monkeypatch.setattr(history, '_open_raw_url', limited)
with ThreadPoolExecutor(max_workers=2) as executor:
first = executor.submit(history._display_version, 'https://api.github.com/repos/a/b', 'a' * 40)
assert entered.wait(timeout=0.8)
second = executor.submit(history._display_version, 'https://api.github.com/repos/a/b', 'b' * 40)
try:
assert queued.wait(timeout=0.8)
finally:
release.set()
for future in (first, second):
with pytest.raises(history.HistoryUnavailable):
future.result(timeout=0.8)
assert counts['http'] == 1
def test_distinct_api_requests_serialize_and_observe_first_quota(history, monkeypatch):
start = threading.Barrier(3)
entered, duplicate, release = threading.Event(), threading.Event(), threading.Event()
calls = []
def limited(request, **kwargs):
calls.append(request.full_url)
entered.set()
if len(calls) > 1:
duplicate.set()
assert release.wait(timeout=0.8)
raise HTTPError(request.full_url, 429, 'Limited', {'Retry-After': '180'}, io.BytesIO())
monkeypatch.setattr(history, '_open_url', limited)
def caller(suffix):
start.wait(timeout=0.8)
return history._get_json('https://api.github.com/repos/a/b/' + suffix)
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(caller, suffix) for suffix in ['branches/main', 'commits']]
start.wait(timeout=0.8)
try:
assert entered.wait(timeout=0.8)
duplicate.wait(timeout=0.1)
finally:
release.set()
for future in futures:
with pytest.raises(history.HistoryUnavailable):
future.result(timeout=0.8)
assert len(calls) == 1
+99 -34
View File
@@ -10,6 +10,7 @@ import subprocess
import threading
import time
from collections import OrderedDict
from contextlib import contextmanager
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from http.client import HTTPException
@@ -31,6 +32,10 @@ _cache = OrderedDict()
_cache_lock = threading.Lock()
_api_backoff_until = 0.0
_api_retry_at = 0.0
_api_request_lock = threading.Lock()
_raw_backoff_until = 0.0
_raw_retry_at = 0.0
_request_locks = {}
VERSION_CACHE_TTL = 6 * 60 * 60
MAX_VERSION_CACHE_ENTRIES = 1024
MAX_VERSION_BYTES = 64 * 1024
@@ -82,8 +87,50 @@ def _is_quota_failure(error):
(headers.get('X-RateLimit-Remaining') == '0' or (wait.isdigit() and len(wait) <= 6)))
@contextmanager
def _request_lock(kind, url):
"""Coalesce identical misses without retaining locks after callers finish."""
key = (kind, url)
with _cache_lock:
entry = _request_locks.setdefault(key, [threading.Lock(), 0])
entry[1] += 1
try:
with entry[0]:
yield
finally:
with _cache_lock:
entry[1] -= 1
if entry[1] == 0:
del _request_locks[key]
def _retry_at(headers):
wait = headers.get('Retry-After', '').strip()
reset = headers.get('X-RateLimit-Reset', '').strip()
if wait.isdigit() and len(wait) <= 6:
return time.time() + int(wait)
if reset.isdigit() and len(reset) <= 12 and int(reset) > time.time():
return float(reset)
return 0.0
def _record_quota_backoff(retry_at, raw=False):
global _api_backoff_until, _api_retry_at, _raw_backoff_until, _raw_retry_at
delay = max(1, retry_at - time.time()) if retry_at else 60
deadline = time.monotonic() + delay
with _cache_lock:
if raw:
if deadline > _raw_backoff_until:
_raw_backoff_until, _raw_retry_at = deadline, retry_at
elif deadline > _api_backoff_until:
_api_backoff_until, _api_retry_at = deadline, retry_at
def _get_display_version(url):
"""Read only the numeric version literal; never import or execute remote code."""
with _cache_lock:
if time.monotonic() < _raw_backoff_until:
raise HistoryUnavailable(_rate_limit_message(_raw_retry_at))
request = Request(url, headers={'User-Agent': 'StarPilot-Galaxy-VersionPicker'})
try:
with _open_raw_url(request, timeout=HTTP_TIMEOUT) as response:
@@ -103,8 +150,11 @@ def _get_display_version(url):
if error.code == 404:
return None # Older builds can predate the display-version file.
if error.code in (403, 429):
failure = HistoryUnavailable if _is_quota_failure(error) else VersionHistoryError
raise failure('GitHub version metadata access is limited; retry later.') from None
if _is_quota_failure(error):
retry_at = _retry_at(error.headers or {})
_record_quota_backoff(retry_at, raw=True)
raise HistoryUnavailable(_rate_limit_message(retry_at)) from None
raise VersionHistoryError('GitHub version metadata access is limited; retry later.') from None
failure = HistoryUnavailable if 500 <= error.code < 600 else VersionHistoryError
raise failure('GitHub version metadata is unavailable; retry later.') from None
except (URLError, TimeoutError, OSError, HTTPException):
@@ -117,14 +167,16 @@ def _display_version(base, sha):
# The validated origin and full SHA make this an immutable public file URL.
repository = base.removeprefix('https://api.github.com/repos/')
url = f'https://raw.githubusercontent.com/{repository}/{sha}/selfdrive/ui/lib/starpilot_version.py'
# Share the limit across requests, and recheck the cache after acquiring a slot.
with _version_slots:
# Duplicate callers share one miss; cached values do not consume download slots.
with _request_lock('raw', url):
with _cache_lock:
entry = _version_cache.get(url)
if entry is not None and time.monotonic() - entry[0] < VERSION_CACHE_TTL:
_version_cache.move_to_end(url)
return entry[1]
value = _get_display_version(url)
with _version_slots:
# Check raw quota inside the slot, including workers queued by other callers.
value = _get_display_version(url)
with _cache_lock:
_version_cache[url] = (time.monotonic(), value)
_version_cache.move_to_end(url)
@@ -145,8 +197,13 @@ def _rate_limit_message(retry_at):
def _get_json(url):
"""Serialize public API requests and recheck quota after waiting for a turn."""
with _api_request_lock:
return _request_json(url)
def _request_json(url):
"""Fetch one bounded public API response. Kept separate for offline tests."""
global _api_backoff_until, _api_retry_at
with _cache_lock:
if time.monotonic() < _api_backoff_until:
raise HistoryUnavailable(_rate_limit_message(_api_retry_at))
@@ -162,21 +219,10 @@ def _get_json(url):
except HTTPError as error:
error.close()
if error.code in (403, 429):
headers = error.headers or {}
wait = headers.get('Retry-After', '').strip()
reset = headers.get('X-RateLimit-Reset', '').strip()
retry_at = 0.0
if wait.isdigit() and len(wait) <= 6:
retry_at = time.time() + int(wait)
elif reset.isdigit() and len(reset) <= 12 and int(reset) > time.time():
retry_at = float(reset)
# Share only a short quota backoff; unrelated 403 responses do not block
# other lookups. Valid cached responses remain available during backoff.
retry_at = _retry_at(error.headers or {})
# Honor the complete server deadline. Unrelated 403s do not block lookups.
if _is_quota_failure(error):
delay = max(1, min(60, retry_at - time.time())) if retry_at else 60
with _cache_lock:
_api_backoff_until = time.monotonic() + delay
_api_retry_at = retry_at
_record_quota_backoff(retry_at)
failure = HistoryUnavailable if _is_quota_failure(error) else VersionHistoryError
raise failure(_rate_limit_message(retry_at)) from None
if error.code == 404:
@@ -190,20 +236,20 @@ def _get_json(url):
def _json(url, fresh=False, ttl=CACHE_TTL):
now = time.monotonic()
if not fresh:
with _request_lock('api', url):
if not fresh:
with _cache_lock:
entry = _cache.get(url)
if entry is not None and time.monotonic() - entry[0] < ttl:
_cache.move_to_end(url)
return copy.deepcopy(entry[1])
value = _get_json(url)
with _cache_lock:
entry = _cache.get(url)
if entry is not None and now - entry[0] < ttl:
_cache.move_to_end(url)
return copy.deepcopy(entry[1])
value = _get_json(url)
with _cache_lock:
_cache[url] = (time.monotonic(), copy.deepcopy(value))
_cache.move_to_end(url)
while len(_cache) > MAX_CACHE_ENTRIES:
_cache.popitem(last=False)
return value
_cache[url] = (time.monotonic(), copy.deepcopy(value))
_cache.move_to_end(url)
while len(_cache) > MAX_CACHE_ENTRIES:
_cache.popitem(last=False)
return value
def _git(repo_path, *args):
@@ -444,7 +490,17 @@ def list_versions(repo_path, branch, page=1, head=None):
return result
def _saved_display_versions(base, branch, page, requested_head, resolved_head):
# A version literal belongs to an immutable SHA. Reuse its validated saved
# value only after the live branch head and requested commit page are checked.
snapshot = _load_history_snapshot(base, branch, page, requested_head)
if snapshot is None or snapshot['head'] != resolved_head:
return {}
return {row['sha']: row['version'] for row in snapshot['commits'] if 'version' in row}
def _list_versions(base, branch, page, pinned_head):
requested_head = pinned_head
current = _head(base, branch)
if pinned_head is not None:
_ancestor(base, pinned_head, current)
@@ -458,9 +514,18 @@ def _list_versions(base, branch, page, pinned_head):
has_more = bool(_block(base, pinned_head, block_page + 2))
selected = rows[offset:end]
if branch.lower() == 'starpilot' and selected:
saved_versions = _saved_display_versions(base, branch, page, requested_head, pinned_head)
pending = []
for index, row in enumerate(selected):
if row['sha'] in saved_versions:
selected[index] = dict(row, version=saved_versions[row['sha']])
else:
pending.append(index)
if not pending:
return {'branch': branch, 'head': pinned_head, 'page': page, 'hasMore': has_more, 'commits': selected}
executor = ThreadPoolExecutor(max_workers=_VERSION_WORKERS)
try:
futures = {executor.submit(_display_version, base, row['sha']): index for index, row in enumerate(selected)}
futures = {executor.submit(_display_version, base, selected[index]['sha']): index for index in pending}
for future in as_completed(futures):
index = futures[future]
selected[index] = dict(selected[index], version=future.result())
@@ -208,6 +208,31 @@ def check_repository_idle(repo):
# Check independently of operation markers, which may be missing or stale.
if git(repo, 'ls-files', '--unmerged'):
raise InstallError('Repository has unmerged index entries; resolve them before installing or restoring')
# These flags can hide working edits from ordinary patches, and the patches
# cannot restore the flags themselves. Leave both source and index untouched.
for entry in git(repo, 'ls-files', '-v', '-z', binary=True).split(b'\0'):
if entry[:1].islower() or entry[:1] == b'S':
name = entry[2:].decode(errors='replace')
raise InstallError(f'Repository uses assume-unchanged or skip-worktree on {name}; save its contents and clear the flag before installing or restoring')
def _check_ignored_collisions(repo, commit):
# Force checkout also replaces ignored files, including file/directory
# collisions. They are deliberately absent from the ordinary recovery tar.
ignored = [name for name in git(repo, 'ls-files', '--others', '--ignored', '--exclude-standard', '-z', binary=True).split(b'\0') if name]
if not ignored:
return
tracked = set(git(repo, 'ls-tree', '-r', '-z', '--name-only', commit, binary=True).split(b'\0')) - {b''}
directories = set()
for name in tracked:
parts = name.split(b'/')
directories.update(b'/'.join(parts[:index]) for index in range(1, len(parts)))
for name in ignored:
parts = name.split(b'/')
if (name in tracked or name in directories or
any(b'/'.join(parts[:index]) in tracked for index in range(1, len(parts)))):
label = name.decode(errors='replace')
raise InstallError(f'Ignored local path conflicts with the selected source: {label}; move or save it outside the checkout before installing or restoring')
def _check_submodules(repo):
@@ -282,6 +307,7 @@ def restore(folder, *, check_parked, restore_data=False):
validate_target(old)
check_parked()
check_repository_idle(repo)
_check_ignored_collisions(repo, old['commit'])
# Validate the whole archive before changing the checkout.
with tarfile.open(folder / 'untracked.tar') as archive:
for member in archive.getmembers():
@@ -337,6 +363,7 @@ def install(repo, target, *, data_root=Path('/data'), check_parked, progress, re
progress(2, 'Checking compatibility', 100, sha[:10])
check_repository_idle(repo)
_check_submodules(repo)
_check_ignored_collisions(repo, sha)
preflight(repo, sha, require_device_binaries, data_root)
check_parked()
progress(3, 'Saving recovery backup', 0, 'Preserving local changes, settings and model statistics')