From 1e1338f6cebc29254dfd27cb356b2148a37072ba Mon Sep 17 00:00:00 2001 From: Jason Wen Date: Wed, 26 Aug 2026 03:09:08 -0400 Subject: [PATCH] models: resume partial chunked downloads and verify silently --- openpilot/sunnypilot/models/manager.py | 26 +++++++++---------- .../models/tests/test_manager_download.py | 21 ++++++++++++++- 2 files changed, 33 insertions(+), 14 deletions(-) diff --git a/openpilot/sunnypilot/models/manager.py b/openpilot/sunnypilot/models/manager.py index 930eddef6e..787f65e3e6 100644 --- a/openpilot/sunnypilot/models/manager.py +++ b/openpilot/sunnypilot/models/manager.py @@ -105,7 +105,7 @@ class ModelManagerSP: # Clean up start time after download completes del self._download_start_times[model.fileName] - async def _download_chunked(self, base_url: str, base_path: str, artifact) -> None: + async def _download_chunked(self, base_url: str, base_path: str, artifact, skip: frozenset[int] | set[int] = frozenset()) -> None: from openpilot.common.file_chunker import get_chunk_name, get_manifest_path num_chunks = len(artifact.chunks) @@ -117,8 +117,11 @@ class ModelManagerSP: # Shared connection saves a TCP+TLS handshake per chunk. # Keep sequential: the link saturates on one stream and Session is not thread-safe. + completed = len(skip) with requests.Session() as session: for i, _ in enumerate(artifact.chunks): + if i in skip: + continue chunk_url = get_chunk_name(base_url, i, num_chunks) chunk_path = get_chunk_name(base_path, i, num_chunks) chunk_downloaded = 0 @@ -132,12 +135,13 @@ class ModelManagerSP: if self._download_interrupted(): raise Exception("Download cancelled") intra = chunk_downloaded / max(chunk_size, 1) - progress = min(99.0, ((i + intra) / num_chunks) * 100) + progress = min(99.0, ((completed + intra) / num_chunks) * 100) artifact.downloadProgress.status = custom.ModelManagerSP.DownloadStatus.downloading artifact.downloadProgress.progress = progress artifact.downloadProgress.eta = self._calculate_eta(artifact.fileName, progress) self._sync_artifact_progress(artifact) self._report_status() + completed += 1 with open(manifest_path, 'w') as f: # noqa: ASYNC230 f.write(str(num_chunks)) @@ -158,21 +162,17 @@ class ModelManagerSP: full_path = os.path.join(destination_path, filename) try: + # verification is silent: publishing it as progress made the bar climb and + # then fall back to zero when the real download started is_cached = False + valid_chunks: set[int] = set() if len(artifact.chunks) > 0: from openpilot.common.file_chunker import get_chunk_name num_chunks = len(artifact.chunks) - chunks_valid = True for i, chunk in enumerate(artifact.chunks): - chunk_path = get_chunk_name(full_path, i, num_chunks) - if not await verify_file(chunk_path, chunk.sha256): - chunks_valid = False - break - artifact.downloadProgress.progress = ((i + 1) / num_chunks) * 100 - self._sync_artifact_progress(artifact) - self._report_status() - if chunks_valid and num_chunks > 0: - is_cached = True + if await verify_file(get_chunk_name(full_path, i, num_chunks), chunk.sha256): + valid_chunks.add(i) + is_cached = len(valid_chunks) == num_chunks else: if await verify_file(full_path, expected_hash): is_cached = True @@ -186,7 +186,7 @@ class ModelManagerSP: return if len(artifact.chunks) > 0: - await self._download_chunked(url, full_path, artifact) + await self._download_chunked(url, full_path, artifact, skip=valid_chunks) from openpilot.common.file_chunker import get_chunk_name for i, chunk in enumerate(artifact.chunks): chunk_path = get_chunk_name(full_path, i, len(artifact.chunks)) diff --git a/openpilot/sunnypilot/models/tests/test_manager_download.py b/openpilot/sunnypilot/models/tests/test_manager_download.py index 2d990afa52..5a9ff631a5 100644 --- a/openpilot/sunnypilot/models/tests/test_manager_download.py +++ b/openpilot/sunnypilot/models/tests/test_manager_download.py @@ -323,7 +323,7 @@ class TestManagerDownload(ManagerDownloadTestBase): self.manager.params.remove.assert_called_once_with("ModelManager_DownloadRef") def test_cached_bundle_cancel_skips_slot_write(self): - """Cancelling must stop an already-on-disk bundle before it is applied to the slot.""" + """A cancel must stop an already-on-disk bundle before it is applied to the slot.""" def body(): artifact = self.make_artifact(chunked=True) base_path = os.path.join(self.dest, artifact.fileName) @@ -341,6 +341,25 @@ class TestManagerDownload(ManagerDownloadTestBase): assert all(os.path.isfile(p) for p in self.chunk_paths(base_path)), "cancel must not delete cached chunks" self.run_with_server(body) + def test_resume_skips_valid_chunks(self): + """A chunk already on disk is kept and not re-downloaded; progress starts above its share.""" + def body(): + artifact = self.make_artifact(chunked=True) + base_path = os.path.join(self.dest, artifact.fileName) + with open(get_chunk_name(base_path, 0, len(CHUNK_BODIES)), 'wb') as f: + f.write(CHUNK_BODIES[0]) + + asyncio.run(self.manager._process_artifact(artifact, self.dest)) + + chunk0_suffix = get_chunk_name('', 0, len(CHUNK_BODIES)) + assert not any(p.endswith(chunk0_suffix) for p in DownloadHandler.request_paths), "valid chunk was re-downloaded" + for i, expected in enumerate(CHUNK_BODIES): + with open(get_chunk_name(base_path, i, len(CHUNK_BODIES)), 'rb') as f: + assert f.read() == expected + assert os.path.isfile(get_manifest_path(base_path)) + assert min(self.reported) >= (1 / len(CHUNK_BODIES)) * 100 - 1, "progress must not restart below the resumed share" + self.run_with_server(body) + def _make_params_with_store(self): params = mock.MagicMock() store = {}