From ae205a318f6189fe557681803ec9517069b5b42d Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Mon, 5 Oct 2026 01:13:54 +0800 Subject: [PATCH] feat(agents): add incremental zstd driver updates --- .github/workflows/agents-release.yml | 20 +- agents/scripts/build_driver_zips.py | 164 ++++++++- .../scripts/driver_release_packages_test.py | 141 ++++++++ crates/dbx-core/tests/agent_service.rs | 5 + crates/dbx-driver-agent/src/agent_manager.rs | 55 +++ .../src/agent_offline_export.rs | 2 + crates/dbx-driver-agent/src/agent_service.rs | 281 +++++++++++++-- crates/dbx-driver-agent/src/driver_delta.rs | 331 ++++++++++++++++++ crates/dbx-driver-agent/src/lib.rs | 1 + 9 files changed, 974 insertions(+), 26 deletions(-) create mode 100644 crates/dbx-driver-agent/src/driver_delta.rs diff --git a/.github/workflows/agents-release.yml b/.github/workflows/agents-release.yml index 88ecf8a304..7134d5498a 100644 --- a/.github/workflows/agents-release.yml +++ b/.github/workflows/agents-release.yml @@ -1631,9 +1631,26 @@ jobs: - name: Verify managed JDBC assets in all offline ZIPs run: node agents/scripts/verify_offline_jdbc_release.mjs release + - name: Fetch previous agent registry for delta patches + env: + GH_TOKEN: ${{ steps.release-token.outputs.token }} + run: | + mkdir -p prev-release + if gh release download agents-latest --repo "$GITHUB_REPOSITORY" \ + --pattern agent-registry.json --dir prev-release --clobber; then + echo "previous agent-registry.json fetched for delta generation" + else + echo "no previous agents-latest registry found; delta patches are skipped" + rm -rf prev-release + fi + - name: Build single-driver packages run: | - python3 agents/scripts/build_driver_zips.py release --cleanup-sources + PREV_ARGS=() + if [ -f prev-release/agent-registry.json ]; then + PREV_ARGS+=(--previous-registry prev-release/agent-registry.json) + fi + python3 agents/scripts/build_driver_zips.py release --cleanup-sources "${PREV_ARGS[@]}" python3 -m json.tool release/agent-registry.json > /dev/null echo "=== final agent-registry.json ===" cat release/agent-registry.json @@ -1830,6 +1847,7 @@ jobs: --recursive \ --exclude "*" \ --include "dbx-agent-*.tar.zst" \ + --include "dbx-agent-*.tar.zst.delta" \ --endpoint-url "$R2_ENDPOINT" echo "Uploaded single-driver packages" diff --git a/agents/scripts/build_driver_zips.py b/agents/scripts/build_driver_zips.py index a152b88249..7388566af6 100644 --- a/agents/scripts/build_driver_zips.py +++ b/agents/scripts/build_driver_zips.py @@ -5,8 +5,10 @@ import hashlib import io import json import subprocess +import sys import tarfile import tempfile +import urllib.request from pathlib import Path from urllib.parse import urlparse @@ -62,13 +64,148 @@ def update_release_artifact(artifact: dict, output: Path) -> None: artifact["format"] = "tar_zstd" -def build_driver_zips(release_dir: Path) -> list[Path]: +# Incremental updates only pay off for packages big enough to matter on slow +# links; small packages ship a full download either way. +MIN_DELTA_FULL_SIZE = 8 * 1024 * 1024 +# A delta larger than this fraction of the full package is not worth hosting: +# clients would rather re-download the full artifact. +MAX_DELTA_RATIO = 0.5 + + +class DeltaGenerator: + """Generates `zstd --patch-from` deltas between two releases of one artifact. + + The previous registry (from the `agents-latest` release) provides each + driver's previously published version and artifact URL. For every artifact + whose version changed, the previous package is downloaded once, the new + package is compressed against it, and — when the delta is small enough to + be worth it — both the `.delta` file and the registry `delta` block are + emitted. Any failure to produce a delta is logged and skipped: the release + must never break because an incremental update could not be built. + """ + + def __init__( + self, + release_dir: Path, + previous_registry: dict | None, + *, + min_full_size: int = MIN_DELTA_FULL_SIZE, + max_ratio: float = MAX_DELTA_RATIO, + verbose: bool = True, + ) -> None: + self.release_dir = release_dir + self.previous_registry = previous_registry or {"drivers": {}} + self.min_full_size = min_full_size + self.max_ratio = max_ratio + self.verbose = verbose + self._downloads: dict[str, Path] = {} + self._download_dir: Path | None = None + + def _previous_driver(self, driver_name: str) -> dict | None: + return self.previous_registry.get("drivers", {}).get(driver_name) + + def _download_previous(self, url: str) -> Path | None: + if url in self._downloads: + return self._downloads[url] + if self._download_dir is None: + self._download_dir = Path(tempfile.mkdtemp(prefix="dbx-delta-prev-")) + target = self._download_dir / artifact_filename(url) + try: + with urllib.request.urlopen(url) as response, target.open("wb") as out: + out.write(response.read()) + except Exception as error: # noqa: BLE001 - any fetch failure just skips the delta + print(f" delta skipped (previous artifact unavailable: {error})", file=sys.stderr) + return None + self._downloads[url] = target + return target + + def maybe_generate( + self, + driver_name: str, + new_version: str, + new_artifact: dict, + previous_artifact: dict | None, + new_package: Path, + suffix: str, + ) -> None: + """Attaches a `delta` block to the freshly packaged `new_artifact`. + + `previous_artifact` is the same artifact slot (jar, or one native + platform) in the previous registry and only supplies the base package + URL; `suffix` distinguishes native platforms in the delta file name. + """ + if not previous_artifact: + return + if new_package.stat().st_size < self.min_full_size: + return + previous_driver = self._previous_driver(driver_name) or {} + previous_version = str(previous_driver.get("version", "")) + if not previous_version or previous_version == new_version: + return + base_url = previous_artifact.get("url", "") + if not base_url: + return + previous_package = self._download_previous(base_url) + if previous_package is None: + return + + stem = f"dbx-agent-{driver_name}-{previous_version}-to-{new_version}{suffix}" + delta_path = self.release_dir / f"{stem}.tar.zst.delta" + try: + subprocess.run( + [ + "zstd", "-q", "--patch-from", str(previous_package), + str(new_package), "-o", str(delta_path), "--force", + ], + check=True, + ) + except (OSError, subprocess.CalledProcessError) as error: + print(f" delta skipped (zstd failed: {error})", file=sys.stderr) + delta_path.unlink(missing_ok=True) + return + + delta_size = delta_path.stat().st_size + if delta_size > self.max_ratio * new_package.stat().st_size: + delta_path.unlink() + if self.verbose: + ratio = delta_size / new_package.stat().st_size + print(f" delta skipped ({ratio:.0%} of full package, above {self.max_ratio:.0%})") + return + + # `new_artifact["url"]` already points at the packaged tar.zst on this + # release; the delta file rides the same release, so only the file name + # needs swapping. + new_artifact["delta"] = { + "base_version": previous_version, + "url": release_url_with_filename(new_artifact["url"], delta_path.name), + "sha256": hashlib.sha256(delta_path.read_bytes()).hexdigest(), + "size": delta_size, + } + if self.verbose: + ratio = delta_size / new_package.stat().st_size + print(f" delta {delta_path.name} ({delta_size} bytes, {ratio:.0%} of full)") + + +def build_driver_zips( + release_dir: Path, + previous_registry: dict | None = None, + *, + delta_min_full_size: int = MIN_DELTA_FULL_SIZE, + delta_max_ratio: float = MAX_DELTA_RATIO, +) -> list[Path]: registry_path = release_dir / "agent-registry.json" registry = json.loads(registry_path.read_text(encoding="utf-8")) + delta_generator = DeltaGenerator( + release_dir, + previous_registry, + min_full_size=delta_min_full_size, + max_ratio=delta_max_ratio, + ) outputs: list[Path] = [] for driver_name, driver in registry.get("drivers", {}).items(): version = driver["version"] + previous_driver = delta_generator._previous_driver(driver_name) jar_artifact = driver.get("jar") if jar_artifact and jar_artifact.get("size", 0) > 0: filename = artifact_filename(jar_artifact["url"]) @@ -86,6 +223,10 @@ def build_driver_zips(release_dir: Path) -> list[Path]: elif not output.is_file(): raise FileExistsError(f"Reusable Java agent package is not a file: {output}") update_release_artifact(jar_artifact, output) + delta_generator.maybe_generate( + driver_name, version, jar_artifact, + (previous_driver or {}).get("jar"), output, "", + ) outputs.append(output) for platform, artifact in driver.get("native", {}).items(): @@ -104,6 +245,10 @@ def build_driver_zips(release_dir: Path) -> list[Path]: elif not output.is_file(): raise FileExistsError(f"Reusable native agent package is not a file: {output}") update_release_artifact(artifact, output) + delta_generator.maybe_generate( + driver_name, version, artifact, + (previous_driver or {}).get("native", {}).get(platform), output, f"-{platform}", + ) outputs.append(output) registry_path.write_text(json.dumps(registry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") @@ -113,7 +258,7 @@ def build_driver_zips(release_dir: Path) -> list[Path]: def remove_raw_driver_artifacts(release_dir: Path) -> list[Path]: removed: list[Path] = [] for path in sorted(release_dir.glob("dbx-agent-*")): - if path.name.endswith(".tar.zst") or not path.is_file(): + if path.name.endswith(".tar.zst") or path.name.endswith(".delta") or not path.is_file(): continue path.unlink() removed.append(path) @@ -124,9 +269,22 @@ def main() -> None: parser = argparse.ArgumentParser(description="Build tar.zst packages for individual DBX agents") parser.add_argument("release_dir", type=Path) parser.add_argument("--cleanup-sources", action="store_true") + parser.add_argument( + "--previous-registry", + type=Path, + default=None, + help=( + "agent-registry.json of the previous agents release; enables " + "incremental delta patches between consecutive driver versions" + ), + ) args = parser.parse_args() - for path in build_driver_zips(args.release_dir): + previous_registry = None + if args.previous_registry and args.previous_registry.is_file(): + previous_registry = json.loads(args.previous_registry.read_text(encoding="utf-8")) + + for path in build_driver_zips(args.release_dir, previous_registry): print(f"Prepared {path.name} ({path.stat().st_size} bytes)") if args.cleanup_sources: for path in remove_raw_driver_artifacts(args.release_dir): diff --git a/agents/scripts/driver_release_packages_test.py b/agents/scripts/driver_release_packages_test.py index 6dc4fe04e3..682cb766ae 100644 --- a/agents/scripts/driver_release_packages_test.py +++ b/agents/scripts/driver_release_packages_test.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 +import hashlib import io import json +import secrets import subprocess import tarfile import tempfile @@ -466,5 +468,144 @@ class DriverReleasePackagesTest(unittest.TestCase): self.assertIn(f"drivers/{kafka_filename}", archive.namelist()) +class DriverDeltaPatchTest(unittest.TestCase): + """Incremental `zstd --patch-from` deltas emitted by build_driver_zips.""" + + @staticmethod + def _incompressible_filler(size: int) -> bytes: + # Deterministic, incompressible shared payload: two packages built from + # it stay ~`size` bytes large while their version differences stay a + # few bytes, so the delta ratio reflects the small real change. + chunks = [] + counter = 0 + while sum(len(chunk) for chunk in chunks) < size: + chunks.append(hashlib.sha256(f"demo-filler-{counter}".encode()).digest()) + counter += 1 + return b"".join(chunks)[:size] + + @staticmethod + def _write_release(release_dir: Path, version: str, jar_payload: bytes) -> None: + jar = release_dir / f"dbx-agent-demo-{version}.jar" + jar.write_bytes(jar_payload) + registry = { + "jres": {"21": {"version": "21", "platforms": {}}}, + "drivers": { + "demo": { + "version": version, + "label": "Demo", + "min_app_version": "0.6.0", + "jre": "21", + "jar": {"url": f"https://example.com/{jar.name}", "size": jar.stat().st_size}, + } + }, + } + (release_dir / "agent-registry.json").write_text(json.dumps(registry), encoding="utf-8") + + @staticmethod + def _zstd_available() -> bool: + try: + subprocess.run(["zstd", "--version"], capture_output=True, check=True) + return True + except (OSError, subprocess.CalledProcessError): + return False + + def _build_previous_release(self, release_dir: Path) -> None: + filler = self._incompressible_filler(96 * 1024) + jar_v1 = b"PK-demo-jar-v1\x00" + filler + b"\x00trailer-v1" + self._write_release(release_dir, "0.1.0", jar_v1) + build_driver_zips(release_dir) + registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8")) + package = release_dir / "dbx-agent-demo-0.1.0.tar.zst" + # Point the previous registry at the local package so the generator + # fetches its base the same way production fetches the previous release. + registry["drivers"]["demo"]["jar"]["url"] = package.resolve().as_uri() + (release_dir / "previous-agent-registry.json").write_text(json.dumps(registry), encoding="utf-8") + + def test_delta_patch_generated_and_registry_annotated(self) -> None: + if not self._zstd_available(): + self.skipTest("zstd CLI is unavailable") + with tempfile.TemporaryDirectory() as temp_dir: + release_dir = Path(temp_dir) + self._build_previous_release(release_dir) + + filler = self._incompressible_filler(96 * 1024) + jar_v2 = b"PK-demo-jar-v2\x00" + filler + b"\x00trailer-v2" + self._write_release(release_dir, "0.2.0", jar_v2) + previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8")) + build_driver_zips(release_dir, previous, delta_min_full_size=0) + + package_v2 = release_dir / "dbx-agent-demo-0.2.0.tar.zst" + delta_path = release_dir / "dbx-agent-demo-0.1.0-to-0.2.0.tar.zst.delta" + self.assertTrue(delta_path.is_file(), "delta file must be written into the release dir") + self.assertLess(delta_path.stat().st_size, package_v2.stat().st_size) + + registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8")) + delta = registry["drivers"]["demo"]["jar"].get("delta") + self.assertIsNotNone(delta, "registry must carry the delta block") + self.assertEqual(delta["base_version"], "0.1.0") + self.assertEqual(delta["size"], delta_path.stat().st_size) + self.assertEqual(delta["sha256"], hashlib.sha256(delta_path.read_bytes()).hexdigest()) + self.assertTrue(delta["url"].endswith(delta_path.name)) + self.assertEqual(delta["url"].rpartition("/")[0], "https://example.com") + + # The delta must reconstruct the exact full package from the base. + reconstructed = release_dir / "reconstructed.tar.zst" + subprocess.run( + [ + "zstd", "-q", "-d", "--patch-from", + str(release_dir / "dbx-agent-demo-0.1.0.tar.zst"), + str(delta_path), "-o", str(reconstructed), "--force", + ], + check=True, + ) + self.assertEqual(reconstructed.read_bytes(), package_v2.read_bytes()) + + def test_delta_skipped_when_ratio_exceeds_threshold(self) -> None: + if not self._zstd_available(): + self.skipTest("zstd CLI is unavailable") + with tempfile.TemporaryDirectory() as temp_dir: + release_dir = Path(temp_dir) + self._build_previous_release(release_dir) + # A completely unrelated payload produces a delta as large as the + # full package; the generator must drop it and leave no block. + self._write_release(release_dir, "0.2.0", secrets.token_bytes(96 * 1024)) + previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8")) + build_driver_zips(release_dir, previous, delta_min_full_size=0, delta_max_ratio=0.5) + + self.assertEqual(list(release_dir.glob("*.delta")), []) + registry = json.loads((release_dir / "agent-registry.json").read_text(encoding="utf-8")) + self.assertIsNone(registry["drivers"]["demo"]["jar"].get("delta")) + + def test_delta_skipped_for_unchanged_version(self) -> None: + if not self._zstd_available(): + self.skipTest("zstd CLI is unavailable") + with tempfile.TemporaryDirectory() as temp_dir: + release_dir = Path(temp_dir) + self._build_previous_release(release_dir) + # Same version in the previous registry: no delta, and the previous + # artifact must not even be fetched (its URL is intentionally dead). + previous = json.loads((release_dir / "previous-agent-registry.json").read_text(encoding="utf-8")) + previous["drivers"]["demo"]["jar"]["url"] = "https://example.invalid/no-such-package.tar.zst" + build_driver_zips(release_dir, previous, delta_min_full_size=0) + self.assertEqual(list(release_dir.glob("*.delta")), []) + + def test_cleanup_keeps_delta_files(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + release_dir = Path(temp_dir) + delta = release_dir / "dbx-agent-demo-0.1.0-to-0.2.0.tar.zst.delta" + delta.write_bytes(b"delta-frame") + raw = release_dir / "dbx-agent-demo-0.2.0.jar" + raw.write_bytes(b"raw") + removed = remove_raw_driver_artifacts(release_dir) + self.assertEqual(removed, [raw]) + self.assertTrue(delta.is_file()) + + def test_release_uploads_delta_files_to_r2(self) -> None: + workflow = (Path(__file__).resolve().parents[2] / ".github/workflows/agents-release.yml").read_text( + encoding="utf-8" + ) + self.assertIn('--include "dbx-agent-*.tar.zst.delta"', workflow) + + if __name__ == "__main__": unittest.main() diff --git a/crates/dbx-core/tests/agent_service.rs b/crates/dbx-core/tests/agent_service.rs index 8aa15b5245..c4704db5bb 100644 --- a/crates/dbx-core/tests/agent_service.rs +++ b/crates/dbx-core/tests/agent_service.rs @@ -128,6 +128,7 @@ fn registry_with_driver(db_type: &str, version: &str, jre: &str) -> AgentRegistr sha256: None, size: 42, format: None, + delta: None, }), native: std::collections::HashMap::new(), }, @@ -158,6 +159,7 @@ fn registry_with_native_driver(db_type: &str, version: &str, jre: &str) -> Agent sha256: None, size: 0, format: None, + delta: None, }), native: [( AgentManager::current_platform().to_string(), @@ -166,6 +168,7 @@ fn registry_with_native_driver(db_type: &str, version: &str, jre: &str) -> Agent sha256: None, size: 42, format: None, + delta: None, }, )] .into_iter() @@ -190,6 +193,7 @@ fn registry_with_foreign_platform_native_driver(db_type: &str, version: &str) -> sha256: None, size: 42, format: None, + delta: None, }, ) }) @@ -209,6 +213,7 @@ fn registry_with_foreign_platform_native_driver(db_type: &str, version: &str) -> sha256: None, size: 0, format: None, + delta: None, }), native, }, diff --git a/crates/dbx-driver-agent/src/agent_manager.rs b/crates/dbx-driver-agent/src/agent_manager.rs index 9477accf69..5848b189f0 100644 --- a/crates/dbx-driver-agent/src/agent_manager.rs +++ b/crates/dbx-driver-agent/src/agent_manager.rs @@ -184,6 +184,42 @@ mod tests { } } + #[test] + fn registry_artifact_parses_optional_delta_and_stays_backward_compatible() { + let with_delta: ArtifactInfo = serde_json::from_str( + r#"{ + "url": "https://example.com/dbx-agent-demo-0.2.0.tar.zst", + "sha256": "52da05589c140cdf5ceba54999422dfc50a1fd23ee19c9f26ee92869ad099d09", + "size": 71239014, + "delta": { + "base_version": "0.1.73", + "url": "https://example.com/dbx-agent-demo-0.1.73-to-0.2.0.tar.zst.delta", + "sha256": "aa05555555c140cdf5feba54999422dfc50a1fd23ee19c9f26ee92869ad099d09", + "size": 4432615 + } + }"#, + ) + .unwrap(); + let delta = with_delta.delta.as_ref().expect("delta parsed"); + assert_eq!(delta.base_version, "0.1.73"); + assert_eq!(delta.size, 4_432_615); + assert!(delta.url.ends_with(".delta")); + // Round-trips through serialization for the offline export path. + let serialized = serde_json::to_value(&with_delta).unwrap(); + assert!(serialized.get("delta").is_some()); + + // Registries published before delta support have no `delta` key. + let without_delta: ArtifactInfo = serde_json::from_str( + r#"{ + "url": "https://example.com/dbx-agent-demo-0.1.73.tar.zst", + "sha256": "52da05589c140cdf5ceba54999422dfc50a1fd23ee19c9f26ee92869ad099d09", + "size": 71239014 + }"#, + ) + .unwrap(); + assert!(without_delta.delta.is_none()); + } + #[test] fn cleanup_pending_jre_removes_stash_dirs_and_persists() { let manager = test_manager("pending-cleanup"); @@ -704,6 +740,25 @@ pub struct ArtifactInfo { pub size: u64, #[serde(default, skip_serializing_if = "Option::is_none")] pub format: Option, + /// Optional incremental update payload: a zstd patch-from frame that + /// reconstructs this artifact from the artifact of `base_version`. Clients + /// without a matching retained base (or on any reconstruct failure) fall + /// back to downloading `url` in full. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub delta: Option, +} + +/// Describes the incremental update payload of an [ArtifactInfo]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DeltaInfo { + /// Version whose downloaded artifact the delta applies onto. + pub base_version: String, + /// URL of the delta frame (a `.delta` file on the same release as the full artifact). + pub url: String, + /// SHA-256 of the delta file itself. + pub sha256: String, + /// Size of the delta file in bytes. + pub size: u64, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] diff --git a/crates/dbx-driver-agent/src/agent_offline_export.rs b/crates/dbx-driver-agent/src/agent_offline_export.rs index 5c234a5de5..4e4d7a0b08 100644 --- a/crates/dbx-driver-agent/src/agent_offline_export.rs +++ b/crates/dbx-driver-agent/src/agent_offline_export.rs @@ -442,6 +442,7 @@ fn build_export_registry(am: &AgentManager, drivers: &[PreparedDriver], jres: &[ sha256: Some(driver.sha256.clone()), size: driver.size, format: None, + delta: None, }; let (jar, native) = match driver.kind { AgentOfflineArtifactKind::Jar => (Some(artifact), HashMap::new()), @@ -473,6 +474,7 @@ fn build_export_registry(am: &AgentManager, drivers: &[PreparedDriver], jres: &[ sha256: Some(jre.sha256.clone()), size: jre.size, format: Some(ArtifactFormat::TarZstd), + delta: None, }, )]), }, diff --git a/crates/dbx-driver-agent/src/agent_service.rs b/crates/dbx-driver-agent/src/agent_service.rs index 4301a42971..e9bf675f5a 100644 --- a/crates/dbx-driver-agent/src/agent_service.rs +++ b/crates/dbx-driver-agent/src/agent_service.rs @@ -1665,18 +1665,14 @@ async fn install_sqlite_worker_from_registry( current, total_drivers, )); - download_with_progress( + obtain_driver_artifact( am, progress, - "driver", source, - &artifact.url, - &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), &driver.version), + artifact, + db_type, + &driver.version, &download_path, - artifact.size, - artifact.sha256.as_deref(), - Some(CacheIdentity::Driver { db_type, version: &driver.version }), - Some(db_type), current, total_drivers, cancellations, @@ -1798,18 +1794,14 @@ async fn install_agent_driver_from_registry( current, total_drivers, )); - download_with_progress( + obtain_driver_artifact( am, progress, - "driver", source, - &artifact.url, - &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), &driver.version), + artifact, + db_type, + &driver.version, &download_path, - artifact.size, - artifact.sha256.as_deref(), - Some(CacheIdentity::Driver { db_type, version: &driver.version }), - Some(db_type), current, total_drivers, cancellations, @@ -1821,6 +1813,9 @@ async fn install_agent_driver_from_registry( std::fs::remove_file(&download_path).ok(); return Err(AGENT_DOWNLOAD_CANCELED_ERROR.to_string()); } + // Retain the artifact bytes for future incremental updates before the + // install consumes (and deletes) the staged download. + retain_driver_delta_base(am, db_type, &driver.version, &download_path); stop_driver_processes_before_replacement(am, db_type).await?; install_downloaded_driver_artifact( &download_path, @@ -2048,6 +2043,131 @@ fn agent_registry_driver<'a>( registry.drivers.get(db_type) } +/// Downloads `artifact` for installation, preferring the registry's +/// incremental delta when a matching base artifact was retained by a previous +/// install of `db_type`. The staged full artifact lands at `dest` either way; +/// every delta failure falls back to a full download. +#[allow(clippy::too_many_arguments)] +async fn obtain_driver_artifact( + am: &AgentManager, + progress: &impl Fn(AgentProgressEvent), + source: DownloadSource, + artifact: &crate::agent_manager::ArtifactInfo, + db_type: &str, + driver_version: &str, + dest: &Path, + current: Option, + total_drivers: Option, + cancellations: &[&AgentInstallCancellation], +) -> Result<(), String> { + if let Some(delta) = artifact.delta.as_ref() { + let base = crate::driver_delta::find_delta_base(&am.download_cache_dir(), db_type, &delta.base_version); + if let Some(base) = base { + match try_delta_artifact( + am, + progress, + source, + artifact, + delta, + &base, + db_type, + driver_version, + dest, + current, + total_drivers, + cancellations, + ) + .await + { + Ok(()) => { + log::info!( + "[driver-delta:{db_type}] applied incremental update {} -> {} (delta {} bytes of a {} byte artifact)", + delta.base_version, + driver_version, + delta.size, + artifact.size + ); + return Ok(()); + } + Err(err) => { + log::warn!( + "[driver-delta:{db_type}] incremental update failed, falling back to full download: {err}" + ); + let _ = std::fs::remove_file(dest); + } + } + } + } + download_with_progress( + am, + progress, + "driver", + source, + &artifact.url, + &r2_path_with_cache_buster(&github_url_to_r2_path(&artifact.url, "driver"), driver_version), + dest, + artifact.size, + artifact.sha256.as_deref(), + Some(CacheIdentity::Driver { db_type, version: driver_version }), + Some(db_type), + current, + total_drivers, + cancellations, + ) + .await +} + +/// Downloads the delta frame and reconstructs the full artifact onto `dest`. +/// The reconstructed bytes are verified against the full artifact's size and +/// SHA-256; a mismatching (e.g. corrupt) base is caught here and reported as a +/// plain error so the caller can fall back to a full download. +#[allow(clippy::too_many_arguments)] +async fn try_delta_artifact( + am: &AgentManager, + progress: &impl Fn(AgentProgressEvent), + source: DownloadSource, + artifact: &crate::agent_manager::ArtifactInfo, + delta: &crate::agent_manager::DeltaInfo, + base: &Path, + db_type: &str, + driver_version: &str, + dest: &Path, + current: Option, + total_drivers: Option, + cancellations: &[&AgentInstallCancellation], +) -> Result<(), String> { + let file_name = dest + .file_name() + .and_then(|value| value.to_str()) + .ok_or_else(|| "Driver artifact has no file name".to_string())?; + let delta_path = dest + .parent() + .ok_or_else(|| "Driver artifact has no parent directory".to_string())? + .join(format!(".{file_name}.delta")); + let cache_buster_version = format!("{driver_version}-from-{}", delta.base_version); + download_with_progress( + am, + progress, + "driver", + source, + &delta.url, + &r2_path_with_cache_buster(&github_url_to_r2_path(&delta.url, "driver"), &cache_buster_version), + &delta_path, + delta.size, + Some(delta.sha256.as_str()), + Some(CacheIdentity::Driver { db_type, version: &cache_buster_version }), + Some(db_type), + current, + total_drivers, + cancellations, + ) + .await?; + let outcome = crate::driver_delta::apply_zstd_delta(base, &delta_path, dest) + .and_then(|()| validate_artifact_integrity(dest, artifact.size, artifact.sha256.as_deref())); + let _ = std::fs::remove_file(&delta_path); + outcome +} + #[allow(clippy::too_many_arguments)] async fn download_with_progress( am: &AgentManager, @@ -2525,11 +2645,28 @@ fn prune_jre_download_cache(am: &AgentManager, jre_key: &str) -> Result<(), Stri } fn cleanup_driver_download_cache_after_success(am: &AgentManager, db_type: &str) { - if let Err(err) = prune_driver_download_cache(am, db_type) { + // Keep the retained delta base (future incremental updates read it back); + // drop the transient download entries for this driver. + let prefix = format!("driver-{}-", cache_file_token(db_type)); + let base_prefix = format!("driver-{}-base-", cache_file_token(db_type)); + if let Err(err) = remove_download_cache_entries( + am, + |name| name.starts_with(&prefix) && !name.starts_with(&base_prefix), + "cached driver download", + ) { log::warn!("Failed to clean cached download for {db_type}: {err}"); } } +/// Retains the just-installed artifact as the delta base for the next +/// incremental update of `db_type`. Best-effort: a missing base only costs a +/// full download later. +fn retain_driver_delta_base(am: &AgentManager, db_type: &str, version: &str, downloaded: &Path) { + if let Err(err) = crate::driver_delta::retain_delta_base(&am.download_cache_dir(), db_type, version, downloaded) { + log::warn!("Failed to retain delta base for {db_type} {version}: {err}"); + } +} + fn cleanup_jre_download_cache_after_success(am: &AgentManager, jre_key: &str) { if let Err(err) = prune_jre_download_cache(am, jre_key) { log::warn!("Failed to clean cached JRE download for {jre_key}: {err}"); @@ -4543,10 +4680,17 @@ mod agent_registry_install_tests { sha256: None, size: 0, format: None, + delta: None, }), native: [( AgentManager::current_platform().to_string(), - ArtifactInfo { url: native_url.to_string(), sha256: None, size: native_size, format: None }, + ArtifactInfo { + url: native_url.to_string(), + sha256: None, + size: native_size, + format: None, + delta: None, + }, )] .into_iter() .collect(), @@ -4564,7 +4708,7 @@ mod agent_registry_install_tests { label: db_type.to_string(), min_app_version: "0.1.0".to_string(), jre: DEFAULT_JRE_KEY.to_string(), - jar: Some(ArtifactInfo { url: url.to_string(), sha256: None, size, format: None }), + jar: Some(ArtifactInfo { url: url.to_string(), sha256: None, size, format: None, delta: None }), native: std::collections::HashMap::new(), }, ); @@ -4809,7 +4953,7 @@ mod agent_registry_install_tests { version: version.to_string(), platforms: [( AgentManager::current_platform().to_string(), - ArtifactInfo { url: url.to_string(), sha256: None, size, format: None }, + ArtifactInfo { url: url.to_string(), sha256: None, size, format: None, delta: None }, )] .into_iter() .collect(), @@ -4951,6 +5095,7 @@ mod agent_registry_install_tests { sha256: None, size: 8, format: None, + delta: None, }, ); cache_test_registry(registry).await; @@ -5145,6 +5290,77 @@ mod agent_registry_install_tests { assert!(cache_path.exists()); } + #[tokio::test] + async fn successful_delta_download_is_removed_by_driver_cache_cleanup() { + let manager = test_manager("delta-cache-cleanup"); + let db_type = "demo"; + let base_version = "0.1.0"; + let driver_version = "0.2.0"; + let delta_url = "https://example.invalid/demo.delta"; + let rebuilt = b"reconstructed driver package"; + let delta_bytes = zstd::stream::encode_all(rebuilt.as_slice(), 3).unwrap(); + let delta_sha256 = format!("{:x}", Sha256::digest(&delta_bytes)); + let artifact_sha256 = format!("{:x}", Sha256::digest(rebuilt)); + + let base_path = manager.download_cache_dir().join(crate::driver_delta::delta_base_file_name( + db_type, + base_version, + "base.tar.zst", + )); + std::fs::create_dir_all(base_path.parent().unwrap()).unwrap(); + std::fs::write(&base_path, b"unused base for a standalone zstd frame").unwrap(); + + let dest = manager.driver_dir(db_type).join(".agent.tar.zst"); + let delta_path = dest.parent().unwrap().join("..agent.tar.zst.delta"); + let cache_version = format!("{driver_version}-from-{base_version}"); + let cache_path = cached_download_path( + &manager, + delta_url, + delta_bytes.len() as u64, + Some(&delta_sha256), + Some(CacheIdentity::Driver { db_type, version: &cache_version }), + &delta_path, + ); + std::fs::write(&cache_path, &delta_bytes).unwrap(); + + let artifact = ArtifactInfo { + url: "https://example.invalid/demo.tar.zst".to_string(), + sha256: Some(artifact_sha256), + size: rebuilt.len() as u64, + format: Some(ArtifactFormat::TarZstd), + delta: None, + }; + let delta = crate::agent_manager::DeltaInfo { + base_version: base_version.to_string(), + url: delta_url.to_string(), + sha256: delta_sha256, + size: delta_bytes.len() as u64, + }; + + try_delta_artifact( + &manager, + &|_| {}, + DownloadSource::Cnb, + &artifact, + &delta, + &base_path, + db_type, + driver_version, + &dest, + None, + None, + &[], + ) + .await + .unwrap(); + + assert_eq!(std::fs::read(&dest).unwrap(), rebuilt); + assert!(cache_path.exists()); + cleanup_driver_download_cache_after_success(&manager, db_type); + assert!(!cache_path.exists()); + assert!(base_path.exists()); + } + #[tokio::test] async fn registry_install_replaces_hive_legacy_jar_with_native_driver() { let manager = test_manager("hive-native-replaces-legacy-jar"); @@ -5197,11 +5413,23 @@ mod agent_registry_install_tests { let mut native = std::collections::HashMap::new(); native.insert( "linux-x64".to_string(), - ArtifactInfo { url: x64_url.to_string(), sha256: None, size: x64_bytes.len() as u64, format: None }, + ArtifactInfo { + url: x64_url.to_string(), + sha256: None, + size: x64_bytes.len() as u64, + format: None, + delta: None, + }, ); native.insert( "linux-aarch64".to_string(), - ArtifactInfo { url: arm_url.to_string(), sha256: None, size: arm_bytes.len() as u64, format: None }, + ArtifactInfo { + url: arm_url.to_string(), + sha256: None, + size: arm_bytes.len() as u64, + format: None, + delta: None, + }, ); let mut drivers = std::collections::HashMap::new(); drivers.insert( @@ -5216,6 +5444,7 @@ mod agent_registry_install_tests { sha256: None, size: 0, format: None, + delta: None, }), native, }, @@ -6387,6 +6616,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&jre_bytes)), size: jre_bytes.len() as u64, format: None, + delta: None, }, )] .into_iter() @@ -6406,6 +6636,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&jar_bytes)), size: jar_bytes.len() as u64, format: None, + delta: None, }), native: std::collections::HashMap::new(), jre: jre_key.to_string(), @@ -6466,6 +6697,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&unrelated_jre_bytes)), size: unrelated_jre_bytes.len() as u64, format: None, + delta: None, }, )] .into_iter() @@ -6485,6 +6717,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&jar_bytes)), size: jar_bytes.len() as u64, format: None, + delta: None, }), native: std::collections::HashMap::new(), jre: "temurin-21".to_string(), @@ -6527,6 +6760,7 @@ mod agent_registry_install_tests { sha256: Some(sha256), size, format: None, + delta: None, }, )] .into_iter() @@ -6571,6 +6805,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&jre_bytes)), size: jre_bytes.len() as u64, format: None, + delta: None, }, )] .into_iter() @@ -6590,6 +6825,7 @@ mod agent_registry_install_tests { sha256: Some(sha256_bytes(&jar_bytes)), size: jar_bytes.len() as u64, format: None, + delta: None, }), native: std::collections::HashMap::new(), jre: jre_key.to_string(), @@ -6666,6 +6902,7 @@ mod agent_registry_install_tests { sha256: None, size: 0, format: None, + delta: None, }, ) }) diff --git a/crates/dbx-driver-agent/src/driver_delta.rs b/crates/dbx-driver-agent/src/driver_delta.rs new file mode 100644 index 0000000000..2e96d46875 --- /dev/null +++ b/crates/dbx-driver-agent/src/driver_delta.rs @@ -0,0 +1,331 @@ +//! Incremental (delta) driver artifact updates. +//! +//! A delta artifact is a zstd frame produced by compressing the new package +//! with the previous package as a raw-content prefix (`zstd --patch-from`). +//! Reconstructing the new package only needs the previously downloaded +//! artifact — retained in the download cache as a "delta base" after every +//! successful install — plus a small delta file. The reconstructed output is +//! always verified against the full artifact's SHA-256 before it is used, and +//! every failure path falls back to a full download, so a missing or corrupt +//! base can never break an install. + +use std::path::{Path, PathBuf}; + +use zstd::zstd_safe::{DCtx, DParameter, InBuffer, OutBuffer}; + +/// `ZSTD_d_windowLogMax` cap for decoding patch-from frames. The window must +/// cover the whole base artifact; 31 (2 GiB) is above any driver package we +/// ship, and zstd only allocates what the frame actually needs. +const DECODE_WINDOW_LOG_MAX: u32 = 31; + +/// Reconstructs the full artifact at `dest_path` by applying the delta frame +/// `delta_path` onto the base artifact `base_path`. +/// +/// Returns an error on any I/O or zstd failure; callers treat errors as a +/// signal to fall back to a full download. If the base does not match the one +/// the delta was produced from, decoding may still "succeed" with garbage +/// output — the caller's SHA-256 check on the result is what catches that. +pub fn apply_zstd_delta(base_path: &Path, delta_path: &Path, dest_path: &Path) -> Result<(), String> { + let base = + std::fs::read(base_path).map_err(|err| format!("Failed to read delta base {}: {err}", base_path.display()))?; + let delta = std::fs::read(delta_path) + .map_err(|err| format!("Failed to read delta artifact {}: {err}", delta_path.display()))?; + + let mut output = Vec::new(); + decode_with_prefix(&base, &delta, &mut output)?; + + if let Some(parent) = dest_path.parent() { + std::fs::create_dir_all(parent).map_err(|err| format!("Failed to create delta output directory: {err}"))?; + } + std::fs::write(dest_path, &output) + .map_err(|err| format!("Failed to write reconstructed artifact {}: {err}", dest_path.display()))?; + Ok(()) +} + +fn decode_with_prefix(prefix: &[u8], frame: &[u8], output: &mut Vec) -> Result<(), String> { + let mut ctx = DCtx::create(); + ctx.set_parameter(DParameter::WindowLogMax(DECODE_WINDOW_LOG_MAX)) + .map_err(|code| format!("zstd decode window configuration failed: {code:?}"))?; + ctx.ref_prefix(prefix).map_err(|code| format!("zstd delta base registration failed: {code:?}"))?; + + let mut input_pos = 0usize; + let mut out_scratch = vec![0_u8; DCtx::out_size().max(64 * 1024)]; + + loop { + let mut src = InBuffer::around(&frame[input_pos..]); + let mut dst = OutBuffer::around(&mut out_scratch); + let code = + ctx.decompress_stream(&mut dst, &mut src).map_err(|code| format!("zstd delta decode failed: {code:?}"))?; + output.extend_from_slice(dst.as_slice()); + input_pos += src.pos(); + if code == 0 && input_pos >= frame.len() { + return Ok(()); + } + if code > 0 && src.pos() == 0 && input_pos >= frame.len() { + return Err("zstd delta frame is truncated".to_string()); + } + } +} + +/// File name of the retained delta base for `db_type` at `version` inside the +/// download cache. The `driver--` prefix keeps bases aligned with the +/// existing per-driver cache pruning (uninstall, manual cache clear), and the +/// `base--` segment lets [find_delta_base] locate a base by version +/// without knowing the full-artifact URL metadata of that release. +pub fn delta_base_file_name(db_type: &str, version: &str, artifact_file_name: &str) -> String { + format!("driver-{}-base-{}-{}", cache_file_token(db_type), cache_file_token(version), artifact_file_name) +} + +/// Retains `downloaded_artifact` as the delta base for future incremental +/// updates of `db_type`, replacing any base retained for an older version. +/// +/// The base is copied (not moved) so a caller that still needs the downloaded +/// artifact for unpacking keeps working; the transient cache entries for the +/// driver are pruned right after by the regular install cleanup, which is +/// taught to preserve the base returned here. +pub fn retain_delta_base( + cache_dir: &Path, + db_type: &str, + version: &str, + downloaded_artifact: &Path, +) -> Result { + let artifact_file_name = downloaded_artifact + .file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| "Delta base artifact has no file name".to_string())?; + let base_name = delta_base_file_name(db_type, version, artifact_file_name); + let base_path = cache_dir.join(&base_name); + + // Drop bases from other versions of this driver first so at most one base + // per driver is retained at any time (bounded disk usage). + let prefix = format!("driver-{}-base-", cache_file_token(db_type)); + if let Ok(entries) = std::fs::read_dir(cache_dir) { + for entry in entries.flatten() { + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + if name.starts_with(&prefix) && name != base_name { + let _ = std::fs::remove_file(entry.path()); + } + } + } + + if let Some(parent) = base_path.parent() { + std::fs::create_dir_all(parent).map_err(|err| format!("Failed to create delta base directory: {err}"))?; + } + // Windows cannot rename onto an existing file; the version-scoped name + // makes collisions unlikely, but remove first to stay atomic-safe. + let _ = std::fs::remove_file(&base_path); + std::fs::copy(downloaded_artifact, &base_path) + .map_err(|err| format!("Failed to retain delta base {}: {err}", base_path.display()))?; + Ok(base_path) +} + +/// Locates the retained delta base for `db_type` whose version matches +/// `base_version`, if one survives in the download cache. +pub fn find_delta_base(cache_dir: &Path, db_type: &str, base_version: &str) -> Option { + let prefix = format!("driver-{}-base-{}-", cache_file_token(db_type), cache_file_token(base_version)); + let entries = std::fs::read_dir(cache_dir).ok()?; + let mut candidates: Vec = entries + .flatten() + .filter(|entry| entry.file_name().to_str().is_some_and(|name| name.starts_with(&prefix))) + .map(|entry| entry.path()) + .collect(); + candidates.sort(); + candidates.into_iter().find(|path| path.is_file()) +} + +/// Reuses the same sanitizing rules as the driver download cache naming in +/// `agent_service` so bases and transient entries share one namespace. +fn cache_file_token(value: &str) -> String { + let token = value + .chars() + .map(|ch| if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' { ch } else { '-' }) + .collect::() + .trim_matches('-') + .to_string(); + if token.is_empty() { + "unknown".to_string() + } else { + token + } +} + +#[cfg(test)] +mod tests { + use super::*; + use zstd::zstd_safe::{CCtx, CParameter}; + + fn compress_with_prefix(prefix: &[u8], content: &[u8]) -> Vec { + let mut ctx = CCtx::create(); + // Mirror `zstd --patch-from`: a window that covers the whole prefix so + // matches may reference any earlier offset of the base artifact. + let window_log = (usize::BITS - prefix.len().leading_zeros()).saturating_sub(1); + ctx.set_parameter(CParameter::WindowLog(window_log.max(10))).unwrap(); + ctx.set_parameter(CParameter::CompressionLevel(3)).unwrap(); + ctx.ref_prefix(prefix).unwrap(); + + let mut src = InBuffer::around(content); + let mut compressed = Vec::new(); + let mut out_scratch = vec![0_u8; CCtx::out_size()]; + while src.pos() < content.len() { + let mut dst = OutBuffer::around(&mut out_scratch); + ctx.compress_stream(&mut dst, &mut src).unwrap(); + compressed.extend_from_slice(dst.as_slice()); + } + loop { + let mut dst = OutBuffer::around(&mut out_scratch); + let remaining = ctx.end_stream(&mut dst).unwrap(); + compressed.extend_from_slice(dst.as_slice()); + if remaining == 0 { + break; + } + } + compressed + } + + fn temp_dir(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("dbx-driver-delta-{name}-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + fn synthetic_artifact(seed: &str, size: usize, mutate_from: usize) -> Vec { + // Deterministic pseudo-content with large shared regions, so prefix + // matching has real work to do instead of compressing from scratch. + let mut data = Vec::with_capacity(size); + let mut state: u64 = seed.bytes().map(|b| b as u64).sum(); + for _ in 0..size { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + data.push((state >> 33) as u8); + } + if mutate_from < size { + for (i, byte) in b"DELTA-MUTATED-PAYLOAD".iter().enumerate() { + let at = mutate_from + i; + if at < size { + data[at] = *byte; + } + } + } + data + } + + #[test] + fn delta_round_trip_reconstructs_new_artifact_bit_for_bit() { + let old = synthetic_artifact("old", 512 * 1024, usize::MAX); + let new = synthetic_artifact("old", 512 * 1024, 300 * 1024); + let frame = compress_with_prefix(&old, &new); + assert!(frame.len() < new.len(), "delta should be smaller than the full artifact"); + + let dir = temp_dir("roundtrip"); + let base = dir.join("base.tar.zst"); + let delta = dir.join("update.delta"); + let out = dir.join("reconstructed.tar.zst"); + std::fs::write(&base, &old).unwrap(); + std::fs::write(&delta, &frame).unwrap(); + + apply_zstd_delta(&base, &delta, &out).unwrap(); + assert_eq!(std::fs::read(&out).unwrap(), new); + } + + #[test] + fn wrong_base_reconstructs_garbage_instead_of_succeeding_cleanly() { + // A base that does not match the delta's production base may decode + // without a zstd error; the output must then differ from the expected + // artifact, which is exactly what the caller's SHA-256 check detects. + let old = synthetic_artifact("old", 256 * 1024, usize::MAX); + let new = synthetic_artifact("old", 256 * 1024, 128 * 1024); + let frame = compress_with_prefix(&old, &new); + let other_base = synthetic_artifact("other", 256 * 1024, usize::MAX); + + let dir = temp_dir("wrong-base"); + let base = dir.join("base.tar.zst"); + let delta = dir.join("update.delta"); + let out = dir.join("reconstructed.tar.zst"); + std::fs::write(&base, &other_base).unwrap(); + std::fs::write(&delta, &frame).unwrap(); + + if apply_zstd_delta(&base, &delta, &out).is_ok() { + let reconstructed = std::fs::read(&out).unwrap(); + assert_ne!(reconstructed, new, "mismatched base must not reproduce the target"); + } + } + + #[test] + fn truncated_delta_is_rejected() { + let old = synthetic_artifact("old", 128 * 1024, usize::MAX); + let new = synthetic_artifact("old", 128 * 1024, 64 * 1024); + let frame = compress_with_prefix(&old, &new); + + let dir = temp_dir("truncated"); + let base = dir.join("base.tar.zst"); + let delta = dir.join("update.delta"); + let out = dir.join("reconstructed.tar.zst"); + std::fs::write(&base, &old).unwrap(); + std::fs::write(&delta, &frame[..frame.len() / 2]).unwrap(); + + assert!(apply_zstd_delta(&base, &delta, &out).is_err()); + } + + #[test] + fn retained_base_is_found_by_version_and_replaced_by_newer_base() { + let dir = temp_dir("retain"); + let artifact_v1 = dir.join("dbx-agent-demo-0.1.1.tar.zst"); + let artifact_v2 = dir.join("dbx-agent-demo-0.1.2.tar.zst"); + std::fs::write(&artifact_v1, b"v1-bytes").unwrap(); + std::fs::write(&artifact_v2, b"v2-bytes").unwrap(); + + let cache = dir.join("cache"); + std::fs::create_dir_all(&cache).unwrap(); + + retain_delta_base(&cache, "demo", "0.1.1", &artifact_v1).unwrap(); + let found = find_delta_base(&cache, "demo", "0.1.1").expect("base for 0.1.1 retained"); + assert_eq!(std::fs::read(&found).unwrap(), b"v1-bytes"); + assert!(find_delta_base(&cache, "demo", "0.1.2").is_none()); + + // Retaining the next version drops the previous base: one base per driver. + retain_delta_base(&cache, "demo", "0.1.2", &artifact_v2).unwrap(); + assert!(find_delta_base(&cache, "demo", "0.1.1").is_none()); + let found = find_delta_base(&cache, "demo", "0.1.2").expect("base for 0.1.2 retained"); + assert_eq!(std::fs::read(&found).unwrap(), b"v2-bytes"); + } + + /// Real-world end-to-end check against published release artifacts. + /// + /// Ignored by default (CI has no network guarantees and the artifacts are + /// ~68 MB); run manually after downloading two consecutive versions of a + /// driver package, e.g. the snowflake pair from `agents-v0.2.129` and + /// `agents-v0.2.130`: + /// + /// ```text + /// DRIVER_DELTA_E2E_BASE= DRIVER_DELTA_E2E_DELTA= \ + /// DRIVER_DELTA_E2E_EXPECTED_SHA256= \ + /// cargo nextest run -p dbx-driver-agent e2e_real --run-ignored=only + /// ``` + #[test] + #[ignore = "requires manually downloaded release artifacts via DRIVER_DELTA_E2E_* env vars"] + fn e2e_real_artifact_delta_reconstruction() { + let Ok(base) = std::env::var("DRIVER_DELTA_E2E_BASE") else { + panic!("DRIVER_DELTA_E2E_BASE (previous full artifact) is required"); + }; + let Ok(delta) = std::env::var("DRIVER_DELTA_E2E_DELTA") else { + panic!("DRIVER_DELTA_E2E_DELTA (delta frame) is required"); + }; + let expected_sha = std::env::var("DRIVER_DELTA_E2E_EXPECTED_SHA256") + .expect("DRIVER_DELTA_E2E_EXPECTED_SHA256 (sha256 of the new full artifact) is required"); + + let dir = temp_dir("e2e-real"); + let out = dir.join("reconstructed.tar.zst"); + apply_zstd_delta(std::path::Path::new(&base), std::path::Path::new(&delta), &out) + .expect("real delta must apply"); + + let actual_sha = { + use sha2::Digest; + let bytes = std::fs::read(&out).unwrap(); + format!("{:x}", sha2::Sha256::digest(&bytes)) + }; + assert_eq!( + actual_sha, expected_sha, + "reconstructed artifact must match the published full artifact bit-for-bit" + ); + } +} diff --git a/crates/dbx-driver-agent/src/lib.rs b/crates/dbx-driver-agent/src/lib.rs index 0d47ab5380..49bc381036 100644 --- a/crates/dbx-driver-agent/src/lib.rs +++ b/crates/dbx-driver-agent/src/lib.rs @@ -15,6 +15,7 @@ pub mod agent_runtime; pub mod agent_service; pub mod backend_error; pub mod database_capabilities; +pub mod driver_delta; pub mod oracle_oci; pub mod db {