server/lib/src/grpc/migration.rs
//! **`MigrationService`** — the four RPCs a UI drives a one-time registry
//! migration with, and the one place in this server that reaches OUT to someone
//! else's Nexus or Artifactory.
//!
//! # Why four RPCs, when `ReplicationService` next door is one
//!
//! `ReplicationService.Replicate` is a single blocking unary call carrying a
//! dry-run flag, and for its job that is exactly right: it copies between two
//! repositories inside this process, so it is local, bounded, and done before the
//! caller's deadline is anywhere near. This is the pattern this service
//! deliberately DEPARTS from, and the reason is the clock.
//!
//! A migration pulls an entire foreign registry across a network. It runs for
//! HOURS. Shaped as one blocking RPC that would mean:
//!
//! * the gRPC deadline — or any proxy between the UI and here — decides when
//! the migration ends, which is a decision neither of them is qualified to
//! make;
//! * a UI with a spinner and no numbers, for hours;
//! * no way to stop it except killing the server, which strands a half-moved
//! registry;
//! * a client disconnect silently abandoning the run with nobody able to say
//! how far it got.
//!
//! So the long operation is split from the call that starts it. `Start` mints a
//! job and returns in milliseconds. `Watch` streams the progress and any number
//! of watchers may join, leave and rejoin. `Cancel` sets a flag the engine reads
//! at every admission. And `Discover` is the cheap preview that comes first.
//!
//! # Everything here is ADMIN-gated, fail-closed
//!
//! Like promotion and replication. `Start` is a bulk write into repositories
//! across this whole server, so the gate is the server-wide one, and a read-only
//! server refuses it before the request body is even read.
//!
//! # The credential is a parameter, not a stored secret
//!
//! `MigrationSource` carries the source registry's username and password. They
//! live in the job for as long as it runs and are never written down, echoed back
//! by any RPC here, or audited. A one-shot move does not need a stored secret;
//! a stored secret is a standing capability against someone else's registry that
//! outlives the reason it was given. Persisting one is a separate decision with
//! its own holder, and it is not made here.
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};
use holger_agent_lib::migrate as engine;
use traits::migrate::{Disposition, MigrationPlan};
use traits::{ConnectorTrait, RemoteRepository};
use super::holger_proto::migration_service_server::MigrationService;
use super::holger_proto::*;
use super::{coded, internal_err, HolgerGrpc};
use crate::audit::{AuditAction, AuditEvent};
use crate::migration::{LocalRouteSink, ProgressRow};
use holger_errcode as ec;
/// Build the source connector named by the request. The ONLY place this server
/// constructs an outbound registry client.
fn source_connector(src: &MigrationSource) -> Result<Arc<dyn ConnectorTrait>, Status> {
if src.base_url.trim().is_empty() {
return Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"migration source base_url is required",
));
}
let user = (!src.username.is_empty()).then(|| src.username.clone());
let pass = (!src.password.is_empty()).then(|| src.password.clone());
let kind = migration_source::Kind::try_from(src.kind)
.unwrap_or(migration_source::Kind::MigrationSourceUnspecified);
let conn: Arc<dyn ConnectorTrait> = match kind {
migration_source::Kind::Nexus => Arc::new(
connector_nexus::NexusConnector::new(src.base_url.clone(), user, pass)
.map_err(|e| internal_err(format!("nexus connector: {e}")))?,
),
migration_source::Kind::Artifactory => Arc::new(
connector_artifactory::ArtifactoryConnector::new(src.base_url.clone(), user, pass)
.map_err(|e| internal_err(format!("artifactory connector: {e}")))?,
),
migration_source::Kind::MigrationSourceUnspecified => {
return Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"migration source kind must be NEXUS or ARTIFACTORY",
))
}
};
Ok(conn)
}
impl HolgerGrpc {
/// This server's repositories, in the shape the shared classifier compares
/// against. The format is reported through `ArtifactFormat::home().config_key`
/// — holger's own operator word — so the comparison happens between two
/// MAPPED formats and Artifactory's `generic` matches holger's `raw` instead
/// of reading as a collision.
fn destination_repositories(&self) -> Vec<RemoteRepository> {
self.routes
.all_repos()
.into_iter()
.map(|(name, backend)| RemoteRepository {
name,
format: backend.format().home().config_key.to_string(),
repo_type: if backend.is_writable() {
"hosted".into()
} else {
"read-only".into()
},
url: None,
})
.collect()
}
}
/// Map an engine repo plan onto the wire.
fn plan_row_to_proto(r: &traits::migrate::RepoPlan) -> MigrationRepoPlan {
let (disposition, refusal_reason) = match &r.disposition {
Disposition::Existing => ("existing", String::new()),
Disposition::Create => ("create", String::new()),
Disposition::Collision { existing_format } => (
"collision",
format!(
"'{}' already exists here as format '{}'; the source calls it '{}'",
r.target_repo, existing_format, r.source_format
),
),
Disposition::UnmappedFormat => (
"unmapped_format",
format!(
"holger has no repository format for source format '{}'",
r.source_format
),
),
};
// What a person must DO before this row can be filled. Only a `create` row
// has one: holger cannot grow a repository while it is running, so a
// migration that "would create" one is really a migration waiting on an
// operator to declare it and restart. The text comes from the one renderer
// in `traits::migrate`, the same one the sink quotes when it refuses at run
// time, so the console cannot describe the fix differently from the terminal.
let declaration = match &r.disposition {
Disposition::Create => traits::migrate::ron_declaration(&r.target_repo, &r.target_format),
_ => String::new(),
};
MigrationRepoPlan {
source_repo: r.source_repo.clone(),
source_format: r.source_format.clone(),
target_repo: r.target_repo.clone(),
target_format: r.target_format.clone(),
disposition: disposition.to_string(),
refusal_reason,
assets: r.assets as i64,
bytes: r.bytes as i64,
declaration,
}
}
fn plan_to_proto_rows(plan: &MigrationPlan) -> Vec<MigrationRepoPlan> {
plan.repos.iter().map(plan_row_to_proto).collect()
}
fn row_to_proto(r: ProgressRow) -> MigrationProgressEvent {
MigrationProgressEvent {
job_id: r.job_id,
repo: r.repo,
done: r.done as i64,
failed: r.failed as i64,
skipped: r.skipped as i64,
total: r.total as i64,
bytes: r.bytes as i64,
repo_finished: r.repo_finished,
job_finished: r.job_finished,
complete: r.complete,
cancelled: r.cancelled,
message: r.message,
}
}
/// How many progress rows may queue toward one watcher. A watcher that cannot
/// keep up is dropped from its stream, never allowed to slow the migration.
const WATCH_BUFFER: usize = 256;
#[tonic::async_trait]
impl MigrationService for Arc<HolgerGrpc> {
async fn discover(
&self,
request: Request<DiscoverMigrationRequest>,
) -> Result<Response<DiscoverMigrationResponse>, Status> {
// Server-wide ADMIN: a migration creates and fills repositories across
// the whole server, so there is no single repository to scope it to.
// Discover reaches out to a foreign registry with a credential, which is
// an action and not merely a read — same gate as Start.
let _identity = self.authorize_admin(&request, "").await?;
let req = request.into_inner();
let src = req.source.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"migration source is required",
)
})?;
let connector = source_connector(&src)?;
let only = (!src.repository.is_empty()).then(|| src.repository.clone());
let destination = self.destination_repositories();
let (plan, _assets) = engine::discover(connector.as_ref(), &destination, only.as_deref())
.await
.map_err(|e| {
super::functional_status(
"holger-grpc/migrate",
"migration_discovered",
false,
&e.to_string(),
);
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::FailedPrecondition,
format!("could not read the migration source: {e}"),
)
})?;
let refusals = match plan.refuse_collisions() {
Ok(()) => Vec::new(),
Err(e) => e.reasons,
};
super::functional_status(
"holger-grpc/migrate",
"migration_discovered",
true,
&format!(
"{} repos, {} would be created",
plan.repos.len(),
plan.to_create().len()
),
);
Ok(Response::new(DiscoverMigrationResponse {
repos: plan_to_proto_rows(&plan),
total_assets: plan.total_assets() as i64,
total_bytes: plan.total_bytes() as i64,
would_run: refusals.is_empty(),
refusals,
// ONE renderer for the preview text, shared with the CLI dry run, so
// the UI and the terminal cannot describe the same plan differently.
report: plan.dry_run_report(),
}))
}
async fn start(
&self,
request: Request<StartMigrationRequest>,
) -> Result<Response<StartMigrationResponse>, Status> {
let source_ip = self.audit_peer(&request);
let identity = self.authorize_admin(&request, "").await?;
let who = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
// SEC-REVIEW 2026-11-01: a migration lands every asset raw in this Holger's
// repositories. With an intake chain nothing lands past it: refused by name.
#[cfg(feature = "intake")]
if let Some(chain) = &self.intake {
return Err(coded(
&ec::INTAKE_RAW_IMPORT_REFUSED,
tonic::Code::PermissionDenied,
format!(
"a migration lands its assets raw, and this Holger runs an intake chain: \
bring them in as a signed bundle through the import door, or upload into \
'{}' and walk them through the chain",
chain.settings().incoming
),
));
}
let req = request.into_inner();
let src = req.source.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"migration source is required",
)
})?;
let connector = source_connector(&src)?;
let only = (!src.repository.is_empty()).then(|| src.repository.clone());
// The plan is computed HERE, before the job exists, so a refusal is an
// RPC error the caller sees immediately rather than a job that starts and
// then dies. Nothing has been written at this point.
let destination = self.destination_repositories();
let (plan, assets) = engine::discover(connector.as_ref(), &destination, only.as_deref())
.await
.map_err(|e| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::FailedPrecondition,
format!("could not read the migration source: {e}"),
)
})?;
if let Err(refused) = plan.refuse_collisions() {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Upload,
"",
&src.base_url,
&source_ip,
400,
0,
)
.with_detail(format!("migration refused: {}", refused.reasons.join("; "))),
);
super::functional_status(
"holger-grpc/migrate",
"migration_started",
false,
"migration plan refused",
);
return Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::FailedPrecondition,
refused.to_string(),
));
}
let job = self.migration_jobs.start();
let job_id = job.id.clone();
let rows = plan_to_proto_rows(&plan);
self.record_audit(
AuditEvent::new(&who, AuditAction::Upload, "", &src.base_url, &source_ip, 202, 0).with_detail(
format!(
"migration {job_id} started from {} ({} repositories, {} assets, {} would be created)",
src.base_url,
plan.repos.len(),
plan.total_assets(),
plan.to_create().len(),
),
),
);
let opts = engine::MigrationOptions {
concurrency: req.concurrency.max(0) as usize,
// Inverted on the wire so the proto3 zero value is the SAFE one: a
// caller who sets nothing gets every published digest verified.
verify_digests: !req.skip_digest_verification,
};
let sink = Arc::new(LocalRouteSink::new(self.routes.clone()));
let cancel = job.cancel.clone();
let state = self.clone();
let emitter = job.clone();
let terminal_job = job.clone();
let jid = job_id.clone();
let endpoint = src.base_url.clone();
// `Start` RETURNS HERE. The migration runs on its own task for as long as
// it takes; nothing the caller does — including hanging up — changes that.
// A32-14, stays a task: the migration job's own run; its job record (jid) is its ledger and its watchers read it.
tokio::spawn(async move {
let jid_for_rows = jid.clone();
let result =
engine::run_migration(connector, sink, &plan, &assets, &opts, cancel, move |p| {
emitter.emit(ProgressRow {
job_id: jid_for_rows.clone(),
repo: p.repo,
done: p.done,
failed: p.failed,
skipped: p.skipped,
total: p.total,
bytes: p.bytes,
repo_finished: p.repo_finished,
job_finished: false,
complete: false,
cancelled: false,
message: p.message,
});
})
.await;
// The terminal row ALWAYS goes out, on every path. A job whose stream
// simply stops is a job a UI must guess about, and a watcher blocked
// forever on a migration that died is worse than a failure it can read.
let (row, status, detail, ok) = match result {
Ok(report) => {
let msg = format!(
"migration {jid} finished: {} moved, {} already present, {} failed, {} bytes, {} repositories created{}",
report.moved,
report.skipped,
report.failed,
report.bytes,
report.repos_created.len(),
if report.cancelled { " (CANCELLED)" } else { "" },
);
let complete = report.is_complete();
(
ProgressRow {
job_id: jid.clone(),
repo: String::new(),
done: report.moved,
failed: report.failed,
skipped: report.skipped,
total: report.moved + report.skipped + report.failed,
bytes: report.bytes,
repo_finished: true,
job_finished: true,
complete,
cancelled: report.cancelled,
message: msg.clone(),
},
if complete { 200 } else { 500 },
msg,
complete,
)
}
Err(e) => {
let msg = format!("migration {jid} failed: {e}");
(
ProgressRow {
job_id: jid.clone(),
repo: String::new(),
done: 0,
failed: 0,
skipped: 0,
total: 0,
bytes: 0,
repo_finished: true,
job_finished: true,
complete: false,
cancelled: false,
message: msg.clone(),
},
500,
msg,
false,
)
}
};
let bytes = row.bytes;
terminal_job.emit(row);
state.record_audit(
AuditEvent::new(
&who,
AuditAction::Upload,
"",
&endpoint,
&source_ip,
status,
bytes,
)
.with_detail(detail),
);
super::functional_status("holger-grpc/migrate", "migration_completed", ok, &jid);
});
super::functional_status("holger-grpc/migrate", "migration_started", true, &job_id);
Ok(Response::new(StartMigrationResponse {
message: format!(
"migration {job_id} started — follow it with Watch, stop it with Cancel. \
Re-running the same migration later is safe: everything already moved is skipped."
),
job_id,
repos: rows,
}))
}
type WatchStream = ReceiverStream<Result<MigrationProgressEvent, Status>>;
async fn watch(
&self,
request: Request<WatchMigrationRequest>,
) -> Result<Response<Self::WatchStream>, Status> {
// An ADMIN READ: following a migration changes nothing, and refusing it
// on a read-only server would make the one surface that explains what the
// box is doing unreachable on exactly the box that is sealed. Same
// reasoning as the audit query plane.
let _identity = self.authorize_admin_read(&request, "").await?;
let job_id = request.into_inner().job_id;
let job = self.migration_jobs.get(&job_id).ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::NotFound,
format!("no migration job '{job_id}'"),
)
})?;
// Replay + subscribe happen together under the job lock, so a row emitted
// while a watcher is joining lands in exactly one of the two.
let (replay, mut rx) = job.subscribe();
let (tx, out) = mpsc::channel(WATCH_BUFFER);
// A32-14, stays a task: one watcher stream of a migration job, ended with its stream.
tokio::spawn(async move {
let mut finished = replay.iter().any(|r| r.job_finished);
for row in replay {
if tx.send(Ok(row_to_proto(row))).await.is_err() {
return; // the watcher hung up; the job does not care
}
}
while !finished {
match rx.recv().await {
Ok(row) => {
finished = row.job_finished;
if tx.send(Ok(row_to_proto(row))).await.is_err() {
return;
}
}
// A watcher too slow to keep up loses rows rather than
// slowing the migration. It is told so instead of silently
// seeing a gap, and the stream continues with the newer
// state — which is the state it wanted.
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
let _ = tx
.send(Err(coded(
&ec::MIGRATION_WATCHER_LAGGED,
tonic::Code::DataLoss,
format!("watcher fell {n} progress rows behind; following from the current state"),
)))
.await;
}
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
});
super::functional_status("holger-grpc/migrate", "migration_watched", true, &job_id);
Ok(Response::new(ReceiverStream::new(out)))
}
async fn cancel(
&self,
request: Request<CancelMigrationRequest>,
) -> Result<Response<CancelMigrationResponse>, Status> {
let source_ip = self.audit_peer(&request);
let identity = self.authorize_admin(&request, "").await?;
let who = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
let job_id = request.into_inner().job_id;
match self.migration_jobs.cancel(&job_id) {
None => Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::NotFound,
format!("no migration job '{job_id}'"),
)),
Some(cancelling) => {
self.record_audit(
AuditEvent::new(&who, AuditAction::Upload, "", &job_id, &source_ip, 200, 0)
.with_detail(if cancelling {
format!("migration {job_id} cancelled by request")
} else {
format!("migration {job_id} was already finished when cancel arrived")
}),
);
super::functional_status(
"holger-grpc/migrate",
"migration_cancelled",
true,
&job_id,
);
Ok(Response::new(CancelMigrationResponse {
cancelling,
message: if cancelling {
format!(
"migration {job_id} is winding down: no new transfers are admitted and \
the ones in flight are allowed to finish. Re-run the migration to resume."
)
} else {
format!("migration {job_id} had already finished — nothing to cancel")
},
}))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exposed::fast_routes::FastRoutes;
use crate::migration::ProgressRow;
use traits::{ArtifactEntry, ArtifactFormat, ArtifactId, RepositoryBackendTrait};
/// A repository that only has to answer `name` / `format` / `is_writable`:
/// these tests are about the SERVICE, not about a backend.
struct Shell {
name: String,
format: ArtifactFormat,
}
#[async_trait::async_trait]
impl RepositoryBackendTrait for Shell {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
self.format.clone()
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
Ok(())
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
Ok(Vec::new())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, Vec::new(), Vec::new()))
}
}
fn grpc_with(repos: Vec<(&str, ArtifactFormat)>) -> Arc<HolgerGrpc> {
let routes: Vec<(String, Arc<dyn RepositoryBackendTrait>)> = repos
.into_iter()
.map(|(n, f)| {
(
n.to_string(),
Arc::new(Shell {
name: n.to_string(),
format: f,
}) as Arc<dyn RepositoryBackendTrait>,
)
})
.collect();
Arc::new(HolgerGrpc::new(FastRoutes::new(routes)))
}
fn nexus_source(base_url: &str) -> MigrationSource {
MigrationSource {
kind: migration_source::Kind::Nexus as i32,
base_url: base_url.into(),
username: "u".into(),
password: "p".into(),
repository: String::new(),
}
}
#[test]
fn the_destination_is_reported_in_holgers_own_format_words() {
let grpc = grpc_with(vec![
("libs", ArtifactFormat::Maven3),
("wheels", ArtifactFormat::Pip),
("drops", ArtifactFormat::Raw),
]);
let mut repos = grpc.destination_repositories();
repos.sort_by(|a, b| a.name.cmp(&b.name));
let pairs: Vec<(String, String)> = repos.into_iter().map(|r| (r.name, r.format)).collect();
assert_eq!(
pairs,
vec![
("drops".to_string(), "raw".to_string()),
("libs".to_string(), "maven3".to_string()),
("wheels".to_string(), "pip".to_string()),
],
"the classifier compares MAPPED formats; reporting the enum name here would \
make every comparison a collision"
);
}
#[tokio::test]
async fn a_source_without_a_url_is_refused_before_any_connection_is_attempted() {
let grpc = grpc_with(vec![("libs", ArtifactFormat::Maven3)]);
let err = MigrationService::discover(
&grpc,
Request::new(DiscoverMigrationRequest {
source: Some(nexus_source("")),
}),
)
.await
.expect_err("no url, no migration");
assert_eq!(err.code(), tonic::Code::InvalidArgument);
assert!(err.message().contains("base_url"), "{}", err.message());
}
#[tokio::test]
async fn a_source_kind_nobody_named_is_refused_rather_than_guessed() {
let grpc = grpc_with(vec![("libs", ArtifactFormat::Maven3)]);
let src = MigrationSource {
kind: migration_source::Kind::MigrationSourceUnspecified as i32,
base_url: "https://nexus.example".into(),
username: String::new(),
password: String::new(),
repository: String::new(),
};
let err = MigrationService::discover(
&grpc,
Request::new(DiscoverMigrationRequest { source: Some(src) }),
)
.await
.expect_err("an unnamed source kind must refuse");
assert_eq!(err.code(), tonic::Code::InvalidArgument);
assert!(err.message().contains("NEXUS"), "{}", err.message());
}
#[tokio::test]
async fn start_without_a_source_refuses_and_mints_no_job() {
let grpc = grpc_with(vec![("libs", ArtifactFormat::Maven3)]);
let err = MigrationService::start(
&grpc,
Request::new(StartMigrationRequest {
source: None,
concurrency: 0,
skip_digest_verification: false,
}),
)
.await
.expect_err("no source, no job");
assert_eq!(err.code(), tonic::Code::InvalidArgument);
}
#[tokio::test]
async fn watching_a_job_that_does_not_exist_is_not_found_not_an_empty_stream() {
let grpc = grpc_with(vec![]);
let err = MigrationService::watch(
&grpc,
Request::new(WatchMigrationRequest {
job_id: "mig-nope".into(),
}),
)
.await
.expect_err("an unknown job is NOT a stream that ends immediately");
assert_eq!(err.code(), tonic::Code::NotFound);
}
#[tokio::test]
async fn cancelling_an_unknown_job_is_not_found_and_a_known_one_winds_down() {
let grpc = grpc_with(vec![]);
let err = MigrationService::cancel(
&grpc,
Request::new(CancelMigrationRequest {
job_id: "mig-nope".into(),
}),
)
.await
.expect_err("unknown job");
assert_eq!(err.code(), tonic::Code::NotFound);
let job = grpc.migration_jobs.start();
let resp = MigrationService::cancel(
&grpc,
Request::new(CancelMigrationRequest {
job_id: job.id.clone(),
}),
)
.await
.expect("cancel ok")
.into_inner();
assert!(resp.cancelling);
assert!(
job.cancel.load(std::sync::atomic::Ordering::Relaxed),
"RED-when-broken: the flag the engine reads at every admission"
);
assert!(
resp.message.contains("in flight are allowed to finish"),
"{}",
resp.message
);
}
#[tokio::test]
async fn watch_replays_what_a_late_watcher_missed_and_ends_on_the_terminal_row() {
use tokio_stream::StreamExt;
let grpc = grpc_with(vec![]);
let job = grpc.migration_jobs.start();
let jid = job.id.clone();
job.emit(ProgressRow {
job_id: jid.clone(),
repo: "libs".into(),
done: 7,
failed: 1,
skipped: 2,
total: 10,
bytes: 700,
repo_finished: true,
job_finished: false,
complete: false,
cancelled: false,
message: "libs: 7 moved".into(),
});
job.emit(ProgressRow {
job_id: jid.clone(),
repo: String::new(),
done: 7,
failed: 1,
skipped: 2,
total: 10,
bytes: 700,
repo_finished: true,
job_finished: true,
complete: false,
cancelled: false,
message: "migration finished".into(),
});
// A watcher arriving AFTER everything has already happened still gets the
// whole picture — which is the case a UI hits every time it reconnects.
let stream = MigrationService::watch(
&grpc,
Request::new(WatchMigrationRequest {
job_id: jid.clone(),
}),
)
.await
.expect("watch ok")
.into_inner();
let rows: Vec<MigrationProgressEvent> =
stream.map(|r| r.expect("no stream error")).collect().await;
assert_eq!(rows.len(), 2, "the repository row and the terminal row");
assert_eq!(rows[0].repo, "libs");
assert_eq!(rows[0].done, 7);
assert_eq!(rows[0].skipped, 2, "the idempotence count reaches the UI");
assert!(rows[1].job_finished);
assert!(
!rows[1].complete,
"a run with a failure is not complete, and says so"
);
// RED-when-broken: drop the terminal row and this stream never ends — a UI
// watching a finished migration would spin forever.
}
#[test]
fn a_collision_row_reaches_the_wire_with_the_reason_spelled_out() {
let source = traits::RemoteRepository {
name: "shared".into(),
format: "maven2".into(),
repo_type: "hosted".into(),
url: None,
};
let existing = vec![traits::RemoteRepository {
name: "shared".into(),
format: "npm".into(),
repo_type: "hosted".into(),
url: None,
}];
let row = plan_row_to_proto(&traits::migrate::plan_one(&source, &existing, 4, 40));
assert_eq!(row.disposition, "collision");
assert!(row.refusal_reason.contains("npm"), "{}", row.refusal_reason);
assert!(
row.refusal_reason.contains("maven2"),
"{}",
row.refusal_reason
);
let create = plan_row_to_proto(&traits::migrate::plan_one(&source, &[], 4, 40));
assert_eq!(create.disposition, "create");
assert_eq!(create.target_format, "maven3");
assert!(create.refusal_reason.is_empty());
assert_eq!(create.assets, 4);
assert_eq!(create.bytes, 40);
}
/// A `create` row is the one an operator has WORK to do about, and the wire
/// now carries that work: the declaration to paste, verbatim from the single
/// renderer. Every other disposition carries nothing, so a console cannot
/// invite a person to declare a repository that already exists.
#[test]
fn a_create_row_carries_the_declaration_and_no_other_row_does() {
let source = traits::RemoteRepository {
name: "libs-release".into(),
format: "maven2".into(),
repo_type: "hosted".into(),
url: None,
};
let create = plan_row_to_proto(&traits::migrate::plan_one(&source, &[], 4, 40));
assert_eq!(create.disposition, "create");
assert_eq!(
create.declaration,
traits::migrate::ron_declaration("libs-release", "maven3"),
"the wire must quote the one renderer, not a second spelling"
);
let existing = vec![traits::RemoteRepository {
name: "libs-release".into(),
format: "maven2".into(),
repo_type: "hosted".into(),
url: None,
}];
let kept = plan_row_to_proto(&traits::migrate::plan_one(&source, &existing, 4, 40));
assert_eq!(kept.disposition, "existing");
assert!(kept.declaration.is_empty(), "{}", kept.declaration);
let clash = vec![traits::RemoteRepository {
name: "libs-release".into(),
format: "npm".into(),
repo_type: "hosted".into(),
url: None,
}];
let collision = plan_row_to_proto(&traits::migrate::plan_one(&source, &clash, 4, 40));
assert_eq!(collision.disposition, "collision");
assert!(
collision.declaration.is_empty(),
"{}",
collision.declaration
);
}
}