agent/lib/src/migrate.rs
//! **The byte mover for a ONE-TIME migration** — Nexus / Artifactory → holger,
//! in a single pass, with no `.znippy` file on a disk in between.
//!
//! [`traits::migrate`] owns the vocabulary (the format map, the creation plan,
//! the collision refusal, the [`MigrationSink`] seam). This module owns the run:
//! it walks the plan repository by repository, downloads each asset from the
//! source, verifies it against the digest the source PUBLISHED, and writes it
//! into the sink — skipping everything the sink already holds.
//!
//! # Why this is not `dump_all_repos_to_znippy` + `push_znippy_to_connector`
//!
//! Those two are the AIRGAP courier: they exist because the two ends are on
//! different networks and the archive is the thing a human carries between them.
//! A migration is the case where both ends are reachable at once, and forcing it
//! through a file on disk costs a full copy of the registry in local storage and
//! turns one operation into two with a manual step between. The courier stays
//! exactly as it is; this is the other route.
//!
//! # It runs once and it finishes
//!
//! There is no schedule, no watermark and no second pass. [`run_migration`]
//! returns a terminal [`MigrationReport`]. A continuous MIRROR is a different
//! thing and must not be grown out of this one.
//!
//! # Interruption is survivable, because sha256 is
//!
//! Every asset is checked against the sink before it is downloaded
//! ([`MigrationSink::already_has`], keyed on path + the source's published
//! `sha256`). So an interrupted migration is resumed by running it again: what
//! landed is skipped, what did not is fetched. Nothing is duplicated, and a
//! completed migration re-run is an all-skips no-op. That is the property
//! `migration_rerun_skips_everything_it_already_moved` pins.
//!
//! # And the bytes are the bytes the source said they were
//!
//! `RemoteAsset::sha256` carries the digest the source registry published. When
//! there is one, the downloaded bytes are hashed and compared BEFORE they are
//! stored. A mismatch is a per-asset failure — it is never written. A migration
//! is precisely the moment a whole registry crosses into a store that will be
//! trusted afterwards, so bytes that do not match what the source claimed must
//! not be the ones that get trusted.
//!
//! Parallelism is `znippy_zoomies::gatling::io::run` (never rayon): a bounded,
//! no-barrier in-flight pool, so the network round-trips overlap while peak
//! memory stays ≈ cap × payload.
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use anyhow::{Context, Result};
use traits::migrate::{Disposition, MigrationPlan, MigrationSink, RepoPlan};
use traits::{ConnectorTrait, RemoteAsset, RemoteRepository};
/// Default in-flight asset transfers. Each is a download + an upload, both
/// latency-bound, so overlapping a handful hides the round-trips without letting
/// peak memory grow past ≈ cap buffered payloads. Same reasoning (and same
/// number) as the agent's `UPLOAD_CONCURRENCY`.
pub const DEFAULT_MIGRATION_CONCURRENCY: usize = 8;
/// Knobs for one migration run.
#[derive(Debug, Clone)]
pub struct MigrationOptions {
/// In-flight asset transfers. `0` ⇒ [`DEFAULT_MIGRATION_CONCURRENCY`].
pub concurrency: usize,
/// Verify downloaded bytes against `RemoteAsset::sha256` when the source
/// published one. ON by default, and turning it off is a decision an
/// operator has to make explicitly.
pub verify_digests: bool,
}
impl Default for MigrationOptions {
fn default() -> Self {
Self {
concurrency: DEFAULT_MIGRATION_CONCURRENCY,
verify_digests: true,
}
}
}
/// What happened to ONE asset.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AssetOutcome {
/// Downloaded, verified, and written into the sink.
Moved { path: String, bytes: u64 },
/// The sink already held it (path + digest) — the idempotence case.
Skipped { path: String },
/// Not moved. `reason` is the failure, verbatim.
Failed { path: String, reason: String },
/// The run was cancelled before this asset was admitted.
Cancelled { path: String },
}
/// A progress row, emitted as a repository advances. This is what
/// `MigrationService.Watch` streams: repo, done, failed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MigrationProgress {
pub repo: String,
/// Assets written into the sink so far, in THIS repository.
pub done: u64,
/// Assets that could not be moved so far, in this repository.
pub failed: u64,
/// Assets the sink already held (idempotence), in this repository.
pub skipped: u64,
/// Assets the source listed for this repository — the denominator.
pub total: u64,
/// Bytes written so far in this repository.
pub bytes: u64,
/// True on the last row for this repository.
pub repo_finished: bool,
/// Free text: what just happened, or why the repository ended.
pub message: String,
}
/// The terminal result of a whole migration. A migration that ran has one of
/// these; a migration that refused has a `PlanRefused` instead.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MigrationReport {
pub repos_created: Vec<String>,
pub moved: u64,
pub skipped: u64,
pub failed: u64,
pub bytes: u64,
/// True when the run stopped because it was cancelled.
pub cancelled: bool,
/// One line per failed asset, so a failure is named and not just counted.
pub failures: Vec<String>,
}
impl MigrationReport {
/// True when the migration is fully accounted for: nothing failed and it was
/// not cut short. A migration that "succeeded" with failures is a migration
/// an operator would wrongly believe is done.
pub fn is_complete(&self) -> bool {
self.failed == 0 && !self.cancelled
}
}
/// Lowercase-hex sha256 — the digest the source publishes and the one the
/// verification compares against.
fn sha256_hex(data: &[u8]) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(data);
h.finalize().iter().map(|b| format!("{b:02x}")).collect()
}
/// **DISCOVER** — the cheap preview a UI shows before anything is committed.
///
/// Lists the source's repositories, counts each one's assets and sums the sizes
/// the LISTING advertised, and classifies every one against the destination's
/// current repository set. It reads no artifact content: `list_assets` is a
/// metadata call, so discovering a terabyte registry costs a handful of
/// requests, not a terabyte.
pub async fn discover(
source: &dyn ConnectorTrait,
destination_repos: &[RemoteRepository],
only_repo: Option<&str>,
) -> Result<(MigrationPlan, HashMap<String, Vec<RemoteAsset>>)> {
let repos = source
.list_repositories()
.await
.context("listing source repositories")?;
let repos: Vec<RemoteRepository> = match only_repo {
Some(f) => repos.into_iter().filter(|r| r.name == f).collect(),
None => repos,
};
let mut plan = MigrationPlan::default();
let mut assets_by_repo: HashMap<String, Vec<RemoteAsset>> = HashMap::with_capacity(repos.len());
for repo in &repos {
let assets = source
.list_assets(&repo.name)
.await
.with_context(|| format!("listing assets of source repository '{}'", repo.name))?;
let bytes: u64 = assets.iter().filter_map(|a| a.size).sum();
plan.repos.push(traits::migrate::plan_one(
repo,
destination_repos,
assets.len() as u64,
bytes,
));
assets_by_repo.insert(repo.name.clone(), assets);
}
Ok((plan, assets_by_repo))
}
/// **RUN** — move the plan, repository by repository.
///
/// Refuses the whole run up front if the plan has any collision or unmapped
/// format, so a refusal costs nothing and a half-migrated registry is not a
/// state this can reach by that route.
///
/// `cancel` is checked at every job admission: setting it stops admitting work
/// and the run winds down to a terminal report with `cancelled = true`. In-flight
/// transfers are allowed to finish, because tearing down a half-written upload is
/// how a store gets a truncated artifact.
pub async fn run_migration(
source: Arc<dyn ConnectorTrait>,
sink: Arc<dyn MigrationSink>,
plan: &MigrationPlan,
assets_by_repo: &HashMap<String, Vec<RemoteAsset>>,
opts: &MigrationOptions,
cancel: Arc<AtomicBool>,
mut on_progress: impl FnMut(MigrationProgress),
) -> Result<MigrationReport> {
plan.refuse_collisions()?;
let cap = if opts.concurrency == 0 {
DEFAULT_MIGRATION_CONCURRENCY
} else {
opts.concurrency
};
let mut report = MigrationReport::default();
for repo_plan in &plan.repos {
if cancel.load(Ordering::Relaxed) {
report.cancelled = true;
break;
}
let empty: Vec<RemoteAsset> = Vec::new();
let assets = assets_by_repo.get(&repo_plan.source_repo).unwrap_or(&empty);
// Repository creation happens HERE, per repository, and only for a row the
// plan classified as Create — never implicitly on a write. A sink that
// cannot create says so by name and the repository is reported failed
// rather than every one of its assets failing separately.
if repo_plan.disposition == Disposition::Create {
match sink
.create_repository(&repo_plan.target_repo, &repo_plan.target_format)
.await
{
Ok(()) => report.repos_created.push(repo_plan.target_repo.clone()),
Err(e) => {
let msg = format!(
"could not create repository '{}' (holger format '{}'): {e}",
repo_plan.target_repo, repo_plan.target_format
);
report.failed += assets.len() as u64;
report.failures.push(msg.clone());
on_progress(MigrationProgress {
repo: repo_plan.target_repo.clone(),
done: 0,
failed: assets.len() as u64,
skipped: 0,
total: assets.len() as u64,
bytes: 0,
repo_finished: true,
message: msg,
});
continue;
}
}
}
let tally = migrate_one_repo(
source.as_ref(),
sink.as_ref(),
repo_plan,
assets,
opts,
cap,
&cancel,
&mut on_progress,
)
.await?;
report.moved += tally.done;
report.skipped += tally.skipped;
report.failed += tally.failed;
report.bytes += tally.bytes;
report.failures.extend(tally.failures);
if cancel.load(Ordering::Relaxed) {
report.cancelled = true;
}
}
Ok(report)
}
/// Move ONE asset: presence check, download, digest verification, store. Split
/// out of [`migrate_one_repo`] so the transfer future has a concrete named type
/// (see the note at its call site).
async fn move_one_asset(
source: &dyn ConnectorTrait,
sink: &dyn MigrationSink,
target: &str,
asset: RemoteAsset,
verify: bool,
cancel: Arc<AtomicBool>,
) -> Result<AssetOutcome> {
if cancel.load(Ordering::Relaxed) {
return Ok(AssetOutcome::Cancelled {
path: asset.path.clone(),
});
}
// IDEMPOTENCE, before a byte is fetched: the sink is asked whether it already
// holds this path as the artifact the source published.
match sink
.already_has(target, &asset.path, asset.sha256.as_deref())
.await
{
Ok(true) => {
return Ok(AssetOutcome::Skipped {
path: asset.path.clone(),
})
}
Ok(false) => {}
Err(e) => {
return Ok(AssetOutcome::Failed {
path: asset.path.clone(),
reason: format!("destination presence check failed: {e}"),
})
}
}
let data = match source.download_asset(&asset).await {
Ok(d) => d,
Err(e) => {
return Ok(AssetOutcome::Failed {
path: asset.path.clone(),
reason: format!("download: {e}"),
})
}
};
// The bytes must be the bytes the source said they were. A mismatch is never
// stored — a migration is exactly the moment a registry becomes the trusted
// one.
if verify {
if let Some(want) = asset.sha256.as_deref() {
let got = sha256_hex(&data);
if !got.eq_ignore_ascii_case(want) {
return Ok(AssetOutcome::Failed {
path: asset.path.clone(),
reason: format!(
"sha256 mismatch: source published {want}, downloaded bytes are {got} — refusing to store"
),
});
}
}
}
let n = data.len() as u64;
match sink.store(target, &asset.path, &data).await {
Ok(()) => Ok(AssetOutcome::Moved {
path: asset.path.clone(),
bytes: n,
}),
Err(e) => Ok(AssetOutcome::Failed {
path: asset.path.clone(),
reason: format!("store: {e}"),
}),
}
}
/// Per-repository tally, folded into the run report.
struct RepoTally {
done: u64,
failed: u64,
skipped: u64,
bytes: u64,
failures: Vec<String>,
}
#[allow(clippy::too_many_arguments)]
async fn migrate_one_repo(
source: &dyn ConnectorTrait,
sink: &dyn MigrationSink,
repo_plan: &RepoPlan,
assets: &[RemoteAsset],
opts: &MigrationOptions,
cap: usize,
cancel: &Arc<AtomicBool>,
on_progress: &mut impl FnMut(MigrationProgress),
) -> Result<RepoTally> {
let total = assets.len() as u64;
let target = repo_plan.target_repo.as_str();
let verify = opts.verify_digests;
// The jobs are pulled LAZILY by the engine, so the cancel check inside runs
// at admission time — a cancel stops new downloads within one slot of
// latency instead of after the whole repository has been queued.
//
// The asset is CLONED into its job rather than borrowed, and that is
// load-bearing rather than careless. An `async fn` returns an `impl Future`
// capturing its input lifetimes; a closure handed a `&RemoteAsset` is
// inferred at one lifetime and cannot be shown to satisfy the higher-ranked
// bound, so the whole run stops being `Send` and the server can no longer
// spawn the migration onto a task — which is the entire point of a job that
// runs for hours. An owned argument has no lifetime to quantify over. The
// clone is a handful of short strings sitting next to a network round-trip
// and a whole artifact payload; it is not where this loop spends anything.
let jobs = assets
.iter()
.cloned()
.map(move |asset| move_one_asset(source, sink, target, asset, verify, cancel.clone()));
let mut tally = RepoTally {
done: 0,
failed: 0,
skipped: 0,
bytes: 0,
failures: Vec::new(),
};
// gatling, never rayon: a bounded no-barrier in-flight pool. `on_result` is
// serial, so the tally and the progress rows need no lock.
znippy_zoomies::gatling::io::run(jobs, cap, |outcome| {
let message = match &outcome {
AssetOutcome::Moved { path, bytes } => {
tally.done += 1;
tally.bytes += bytes;
format!("moved {path}")
}
AssetOutcome::Skipped { path } => {
tally.skipped += 1;
format!("already present, skipped {path}")
}
AssetOutcome::Failed { path, reason } => {
tally.failed += 1;
let line = format!("{target}/{path}: {reason}");
tally.failures.push(line.clone());
line
}
AssetOutcome::Cancelled { path } => format!("cancelled before {path}"),
};
on_progress(MigrationProgress {
repo: target.to_string(),
done: tally.done,
failed: tally.failed,
skipped: tally.skipped,
total,
bytes: tally.bytes,
repo_finished: false,
message,
});
Ok(())
})
.await?;
on_progress(MigrationProgress {
repo: target.to_string(),
done: tally.done,
failed: tally.failed,
skipped: tally.skipped,
total,
bytes: tally.bytes,
repo_finished: true,
message: format!(
"{}: {} moved, {} already present, {} failed",
target, tally.done, tally.skipped, tally.failed
),
});
Ok(tally)
}
/// A [`MigrationSink`] over a `ConnectorTrait` — the CLI's destination, a holger
/// server reached over the wire.
///
/// Presence is answered from the destination's OWN asset listing, fetched once
/// per repository and cached: one metadata call replaces one HEAD per asset, and
/// a registry with a hundred thousand artifacts makes that the difference between
/// a resumable migration and an unusable one. The cache is deliberately taken
/// ONCE at the start of a repository: a migration is a one-time move into a
/// destination nobody else is writing to, and re-listing per asset would only
/// discover this run's own writes.
pub struct ConnectorSink {
connector: Arc<dyn ConnectorTrait>,
/// repository → (asset path → the digest the destination published, if any).
cache: tokio::sync::Mutex<HashMap<String, HashMap<String, Option<String>>>>,
}
impl ConnectorSink {
pub fn new(connector: Arc<dyn ConnectorTrait>) -> Self {
Self {
connector,
cache: tokio::sync::Mutex::new(HashMap::new()),
}
}
}
#[async_trait::async_trait]
impl MigrationSink for ConnectorSink {
async fn already_has(
&self,
repository: &str,
path: &str,
want_sha256: Option<&str>,
) -> Result<bool> {
let mut cache = self.cache.lock().await;
if !cache.contains_key(repository) {
// A repository the destination does not have yet lists nothing — that
// is an empty map, not an error: the plan already said it is new.
let listed = self
.connector
.list_assets(repository)
.await
.unwrap_or_default();
let map = listed
.into_iter()
.map(|a| (a.path, a.sha256))
.collect::<HashMap<String, Option<String>>>();
cache.insert(repository.to_string(), map);
}
let map = &cache[repository];
let Some(have) = map.get(path) else {
return Ok(false);
};
Ok(match (have.as_deref(), want_sha256) {
// Both ends published a digest: present only when they agree. A
// different digest at the same path is a DIFFERENT artifact, and the
// migration converges the destination onto the source.
(Some(h), Some(w)) => h.eq_ignore_ascii_case(w),
// One end has no digest. The path is there and nothing contradicts it;
// re-uploading identical bytes over it would be waste with no gain.
_ => true,
})
}
async fn store(&self, repository: &str, path: &str, data: &[u8]) -> Result<()> {
self.connector.upload_asset(repository, path, data).await
}
async fn create_repository(&self, repository: &str, format: &str) -> Result<()> {
// A holger reached over `ConnectorTrait` has no create-repository verb:
// the route table is wired from config at boot and is immutable for the
// life of the process. Refuse ONCE with the declaration to add, rather
// than letting every asset fail separately with a 404. The words are
// rendered in `traits::migrate` so this sink and the server-side sink
// cannot tell an operator two different things.
anyhow::bail!(
"{}",
traits::migrate::cannot_create_repository(repository, format)
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use traits::migrate::plan_one;
/// A source registry in a HashMap: repositories, their assets, and the bytes.
struct FakeSource {
repos: Vec<RemoteRepository>,
assets: HashMap<String, Vec<RemoteAsset>>,
blobs: HashMap<String, Vec<u8>>,
/// Paths whose download must fail, to prove a failure is named not hidden.
broken: Vec<String>,
}
#[async_trait::async_trait]
impl ConnectorTrait for FakeSource {
fn name(&self) -> &str {
"fake-source"
}
async fn list_repositories(&self) -> Result<Vec<RemoteRepository>> {
Ok(self.repos.clone())
}
async fn list_assets(&self, repository: &str) -> Result<Vec<RemoteAsset>> {
Ok(self.assets.get(repository).cloned().unwrap_or_default())
}
async fn download_asset(&self, asset: &RemoteAsset) -> Result<Vec<u8>> {
if self.broken.contains(&asset.path) {
anyhow::bail!("upstream said 500");
}
self.blobs
.get(&asset.path)
.cloned()
.ok_or_else(|| anyhow::anyhow!("no such blob {}", asset.path))
}
async fn upload_asset(&self, _r: &str, _p: &str, _d: &[u8]) -> Result<()> {
anyhow::bail!("the source is never written to")
}
}
/// A destination that remembers what it was told to store — and counts the
/// stores, so "did not duplicate" is a measurement and not a hope.
#[derive(Default)]
struct FakeSink {
stored: Mutex<HashMap<(String, String), Vec<u8>>>,
store_calls: Mutex<Vec<String>>,
created: Mutex<Vec<(String, String)>>,
refuse_create: bool,
}
#[async_trait::async_trait]
impl MigrationSink for FakeSink {
async fn already_has(&self, repo: &str, path: &str, want: Option<&str>) -> Result<bool> {
let stored = self.stored.lock().unwrap();
match stored.get(&(repo.to_string(), path.to_string())) {
None => Ok(false),
Some(bytes) => Ok(match want {
Some(w) => sha256_hex(bytes).eq_ignore_ascii_case(w),
None => true,
}),
}
}
async fn store(&self, repo: &str, path: &str, data: &[u8]) -> Result<()> {
self.store_calls
.lock()
.unwrap()
.push(format!("{repo}/{path}"));
self.stored
.lock()
.unwrap()
.insert((repo.to_string(), path.to_string()), data.to_vec());
Ok(())
}
async fn create_repository(&self, repo: &str, format: &str) -> Result<()> {
if self.refuse_create {
anyhow::bail!("this sink creates nothing");
}
self.created
.lock()
.unwrap()
.push((repo.to_string(), format.to_string()));
Ok(())
}
}
fn asset(path: &str, body: &[u8], with_digest: bool) -> RemoteAsset {
RemoteAsset {
path: path.to_string(),
download_url: format!("https://src/{path}"),
content_type: None,
size: Some(body.len() as u64),
sha256: with_digest.then(|| sha256_hex(body)),
change_token: None,
}
}
fn one_repo_source(broken: Vec<String>) -> (FakeSource, Vec<RemoteAsset>) {
let a = b"crate-a-bytes".to_vec();
let b = b"crate-b-bytes".to_vec();
let assets = vec![
asset("a/a-1.0.crate", &a, true),
asset("b/b-2.0.crate", &b, true),
];
let mut blobs = HashMap::new();
blobs.insert("a/a-1.0.crate".to_string(), a);
blobs.insert("b/b-2.0.crate".to_string(), b);
let mut by_repo = HashMap::new();
by_repo.insert("crates".to_string(), assets.clone());
(
FakeSource {
repos: vec![RemoteRepository {
name: "crates".into(),
format: "cargo".into(),
repo_type: "hosted".into(),
url: None,
}],
assets: by_repo,
blobs,
broken,
},
assets,
)
}
#[tokio::test]
async fn discover_counts_assets_and_bytes_without_reading_one_artifact() {
let (src, _) = one_repo_source(vec![]);
let (plan, assets) = discover(&src, &[], None).await.expect("discover");
assert_eq!(plan.repos.len(), 1);
assert_eq!(plan.total_assets(), 2);
assert_eq!(
plan.total_bytes(),
26,
"sizes come from the LISTING, not a download"
);
assert_eq!(
plan.to_create().len(),
1,
"an empty destination means both repos are new"
);
assert_eq!(assets["crates"].len(), 2);
// RED-when-broken: the fake source fails every download_asset for a broken
// path; `discover` calls none, which is what makes it cheap.
}
#[tokio::test]
async fn a_migration_moves_every_asset_and_creates_the_repository() {
let (src, _) = one_repo_source(vec![]);
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let report = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.expect("migration ran");
assert_eq!(report.moved, 2);
assert_eq!(report.skipped, 0);
assert_eq!(report.failed, 0);
assert!(report.is_complete());
assert_eq!(report.repos_created, vec!["crates".to_string()]);
assert_eq!(
sink.created.lock().unwrap().clone(),
vec![("crates".to_string(), "rust".to_string())],
"the created repository carries holger's mapped format, not the source's word"
);
assert_eq!(sink.stored.lock().unwrap().len(), 2);
}
#[tokio::test]
async fn migration_rerun_skips_everything_it_already_moved() {
// THE IDEMPOTENCE PROOF. Run the same migration twice against the same
// sink; the second run must store NOTHING.
let (src1, _) = one_repo_source(vec![]);
let (plan1, assets1) = discover(&src1, &[], None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let first = run_migration(
Arc::new(src1),
sink.clone(),
&plan1,
&assets1,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(first.moved, 2);
let after_first = sink.store_calls.lock().unwrap().len();
assert_eq!(after_first, 2);
// Second pass: the destination now HAS the repository, so the plan reads
// it as Existing and nothing is created either.
let (src2, _) = one_repo_source(vec![]);
let existing = vec![RemoteRepository {
name: "crates".into(),
format: "rust".into(),
repo_type: "hosted".into(),
url: None,
}];
let (plan2, assets2) = discover(&src2, &existing, None).await.unwrap();
assert_eq!(plan2.to_create().len(), 0, "the repository exists now");
let second = run_migration(
Arc::new(src2),
sink.clone(),
&plan2,
&assets2,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(second.moved, 0, "a re-run moves nothing");
assert_eq!(second.skipped, 2, "it recognises both assets by sha256");
assert_eq!(second.failed, 0);
assert_eq!(
sink.store_calls.lock().unwrap().len(),
after_first,
"RED-when-broken: drop the already_has check and this is 4 — every artifact stored twice"
);
assert_eq!(
sink.stored.lock().unwrap().len(),
2,
"and the destination holds 2, not 4"
);
}
#[tokio::test]
async fn an_interrupted_migration_resumes_instead_of_starting_over() {
// Half the registry landed; the rest did not. Re-running moves only the
// rest — which is what makes a multi-hour migration survivable.
let (src, assets_list) = one_repo_source(vec![]);
let sink = Arc::new(FakeSink::default());
// Pretend the first asset landed before the interruption.
let first = &assets_list[0];
sink.store("crates", &first.path, b"crate-a-bytes")
.await
.unwrap();
sink.store_calls.lock().unwrap().clear();
let existing = vec![RemoteRepository {
name: "crates".into(),
format: "rust".into(),
repo_type: "hosted".into(),
url: None,
}];
let (plan, assets) = discover(&src, &existing, None).await.unwrap();
let report = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(
report.skipped, 1,
"the asset that already landed is recognised"
);
assert_eq!(report.moved, 1, "only the missing one is fetched");
assert_eq!(
sink.store_calls.lock().unwrap().clone(),
vec!["crates/b/b-2.0.crate".to_string()]
);
}
#[tokio::test]
async fn bytes_that_do_not_match_the_published_digest_are_never_stored() {
let (mut src, _) = one_repo_source(vec![]);
// The source PUBLISHES one digest and serves different bytes — a
// compromised or corrupted origin.
src.blobs
.insert("a/a-1.0.crate".into(), b"SUBSTITUTED".to_vec());
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let report = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(report.failed, 1);
assert_eq!(report.moved, 1, "the honest asset still moves");
assert!(
!report.is_complete(),
"a run with a digest failure is NOT complete"
);
assert!(
report.failures[0].contains("sha256 mismatch"),
"{:?}",
report.failures
);
assert!(
!sink
.stored
.lock()
.unwrap()
.contains_key(&("crates".into(), "a/a-1.0.crate".into())),
"RED-when-broken: drop the verify and the substituted bytes are stored and trusted"
);
}
#[tokio::test]
async fn a_download_failure_is_named_counted_and_does_not_abort_the_repository() {
let (src, _) = one_repo_source(vec!["a/a-1.0.crate".to_string()]);
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let report = run_migration(
Arc::new(src),
sink,
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(report.failed, 1);
assert_eq!(report.moved, 1, "one bad asset does not abort the other");
assert!(
report.failures[0].contains("upstream said 500"),
"{:?}",
report.failures
);
}
#[tokio::test]
async fn a_collision_refuses_the_run_before_a_single_byte_moves() {
let (src, _) = one_repo_source(vec![]);
let existing = vec![RemoteRepository {
name: "crates".into(),
format: "npm".into(),
repo_type: "hosted".into(),
url: None,
}];
let (plan, assets) = discover(&src, &existing, None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let err = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.expect_err("a collision must refuse");
assert!(err.to_string().contains("different format"), "{err}");
assert!(sink.store_calls.lock().unwrap().is_empty(), "nothing moved");
assert!(sink.created.lock().unwrap().is_empty(), "nothing created");
}
#[tokio::test]
async fn a_sink_that_cannot_create_fails_the_repository_once_not_every_asset() {
let (src, _) = one_repo_source(vec![]);
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let sink = Arc::new(FakeSink {
refuse_create: true,
..Default::default()
});
let report = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
|_| {},
)
.await
.unwrap();
assert_eq!(
report.failures.len(),
1,
"ONE named refusal, not one per asset"
);
assert!(report.failures[0].contains("could not create repository 'crates'"));
assert_eq!(
report.failed, 2,
"but the assets it would have held are still counted lost"
);
assert!(sink.store_calls.lock().unwrap().is_empty());
}
#[tokio::test]
async fn cancel_stops_admitting_work_and_the_report_says_it_was_cancelled() {
let (src, _) = one_repo_source(vec![]);
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let sink = Arc::new(FakeSink::default());
let cancel = Arc::new(AtomicBool::new(true)); // cancelled before it starts
let report = run_migration(
Arc::new(src),
sink.clone(),
&plan,
&assets,
&MigrationOptions::default(),
cancel,
|_| {},
)
.await
.unwrap();
assert!(report.cancelled);
assert_eq!(report.moved, 0);
assert!(sink.store_calls.lock().unwrap().is_empty());
assert!(
!report.is_complete(),
"a cancelled migration is not a complete one"
);
}
#[tokio::test]
async fn progress_rows_name_the_repository_and_end_with_a_finished_row() {
let (src, _) = one_repo_source(vec![]);
let (plan, assets) = discover(&src, &[], None).await.unwrap();
let rows = Arc::new(Mutex::new(Vec::new()));
let sink = Arc::new(FakeSink::default());
let collect = rows.clone();
run_migration(
Arc::new(src),
sink,
&plan,
&assets,
&MigrationOptions::default(),
Arc::new(AtomicBool::new(false)),
move |p| collect.lock().unwrap().push(p),
)
.await
.unwrap();
let rows = rows.lock().unwrap();
assert_eq!(rows.len(), 3, "two asset rows and one finished row");
assert!(rows.iter().all(|r| r.repo == "crates"));
assert!(rows.iter().all(|r| r.total == 2));
let last = rows.last().unwrap();
assert!(last.repo_finished);
assert_eq!(last.done, 2);
assert!(last.message.contains("2 moved"));
}
#[tokio::test]
async fn the_connector_sink_refuses_to_create_and_hands_over_the_declaration() {
let (src, _) = one_repo_source(vec![]);
let sink = ConnectorSink::new(Arc::new(src));
let err = sink
.create_repository("libs", "maven3")
.await
.expect_err("cannot create");
let text = err.to_string();
assert!(text.contains("ron_name: \"libs\""), "{text}");
assert!(text.contains("ron_repo_type: \"maven3\""), "{text}");
assert!(
text.contains("dry run"),
"it must point at the preview: {text}"
);
}
}