server/lib/src/migration.rs
//! **MIGRATION, server side** — the incumbent registry moving into THIS holger,
//! driven by a UI instead of a terminal.
//!
//! The engine is not here. The classifier is `traits::migrate`, the byte mover is
//! `holger_agent_lib::migrate`, and both are shared verbatim with the
//! `holger-agent --from nexus --to holger` CLI route. What lives here is the two
//! things only a server has:
//!
//! 1. [`LocalRouteSink`] — a [`MigrationSink`] that writes into this server's own
//! repositories instead of back out over the network. A migration into the box
//! you are running should not make an HTTP round trip to itself.
//! 2. [`MigrationJobs`] — the ledger that makes `Start` able to return in
//! milliseconds while the work runs for hours: one entry per job, a cancel
//! flag, a live progress broadcast, and a replay of the latest row per
//! repository so a UI that reconnects is not blind.
//!
//! # Why a job and not a blocking RPC
//!
//! `ReplicationService.Replicate` next door is one blocking unary call with a
//! dry-run flag, and for what it does — copying between two repositories inside
//! this process — that is the right shape. A migration pulls someone else's whole
//! registry across a network and runs for hours. A single blocking RPC over that
//! hands the decision of when the migration ends to a gRPC deadline or to
//! whatever proxy is in the middle, leaves a UI with a spinner and no numbers,
//! offers no way to stop it short of killing the server, and turns a client
//! disconnect into a half-moved registry nobody can account for. So `Start`
//! returns a job id and nothing else, and the progress is a stream.
//!
//! # ONE TIME
//!
//! A migration finishes. There is no schedule, no watermark and no second pass
//! here, and a continuous mirror must not be grown out of it.
//!
//! # The credential is a parameter
//!
//! The source's username/password arrive in the request, live in the job for as
//! long as it runs, and are never written down, echoed back or audited. A
//! one-shot move does not need a stored secret, and 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.
use crate::lock::LockOrRecover as _;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use traits::migrate::MigrationSink;
use crate::exposed::fast_routes::FastRoutes;
/// A [`MigrationSink`] over this server's OWN route table.
///
/// Presence and storage both go through `RepositoryBackendTrait::handle_http2_request`,
/// the path-keyed door every ecosystem backend already implements — the same door
/// a `cargo publish` or a `pip upload` arrives at. That matters: a Nexus asset is
/// identified by a repository-relative PATH, and turning a path back into an
/// `ArtifactId` is format-specific and not reliably invertible. The path door is
/// the seam that speaks the source's vocabulary without inventing a translation.
///
/// Every backend call is synchronous and does file I/O, so each one is wrapped in
/// `spawn_blocking`: a migration saturating the async reactor with blocking reads
/// would stall every other request the server is serving while it runs.
pub struct LocalRouteSink {
routes: FastRoutes,
}
impl LocalRouteSink {
pub fn new(routes: FastRoutes) -> Self {
Self { routes }
}
/// `/{repo}/{path}` — the sub-URL shape every backend's HTTP door parses
/// (it strips its own repository name off the front).
fn suburl(repository: &str, path: &str) -> String {
format!("/{}/{}", repository, path.trim_start_matches('/'))
}
}
#[async_trait::async_trait]
impl MigrationSink for LocalRouteSink {
async fn already_has(
&self,
repository: &str,
path: &str,
want_sha256: Option<&str>,
) -> anyhow::Result<bool> {
let Some(backend) = self.routes.lookup(repository) else {
// The repository is not (yet) served. Not an error: the plan already
// classified it, and "absent" simply means nothing is present.
return Ok(false);
};
let suburl = Self::suburl(repository, path);
let (status, _headers, bytes) =
tokio::task::spawn_blocking(move || backend.handle_http2_request("GET", &suburl, &[]))
.await
.map_err(|e| anyhow::anyhow!("presence check task failed: {e}"))??;
if status != 200 {
return Ok(false);
}
Ok(match want_sha256 {
// The source published a digest: present only when the bytes this
// server already holds ARE that artifact. Different bytes at the same
// path are a different artifact, and the migration converges onto the
// source.
Some(want) => nornir_hash::sha256_hex(&bytes).eq_ignore_ascii_case(want),
// No digest to compare. The path is served and nothing contradicts
// it; re-fetching the artifact would be waste with no gain.
None => true,
})
}
async fn store(&self, repository: &str, path: &str, data: &[u8]) -> anyhow::Result<()> {
let backend = self.routes.lookup(repository).ok_or_else(|| {
anyhow::anyhow!("repository '{repository}' is not served by this server")
})?;
if !backend.is_writable() {
anyhow::bail!("repository '{repository}' is read-only on this server");
}
let suburl = Self::suburl(repository, path);
let body = data.to_vec();
let (status, _headers, response) = tokio::task::spawn_blocking(move || {
backend.handle_http2_request("PUT", &suburl, &body)
})
.await
.map_err(|e| anyhow::anyhow!("store task failed: {e}"))??;
if (200..300).contains(&status) {
Ok(())
} else {
anyhow::bail!(
"storing '{path}' in '{repository}' answered HTTP {status}: {}",
String::from_utf8_lossy(&response)
)
}
}
async fn create_repository(&self, repository: &str, format: &str) -> anyhow::Result<()> {
// Same refusal, same words, as the CLI's `ConnectorSink` — rendered once
// in `traits::migrate`. holger wires its route table from config at boot
// (`FastRoutes`, a first-byte bucketed array the hot path reads without a
// lock, built from raw pointers into the wired config) and it is
// immutable thereafter, so no running server can grow a repository
// mid-migration. The plan named every repository that must exist; a dry
// run prints them all, so the whole set is declared in one pass.
anyhow::bail!(
"{}",
traits::migrate::cannot_create_repository(repository, format)
)
}
}
/// One progress row, as the ledger keeps it. The gRPC layer maps this onto
/// `MigrationProgressEvent`; nothing in this module knows about proto.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProgressRow {
pub job_id: String,
pub repo: String,
pub done: u64,
pub failed: u64,
pub skipped: u64,
pub total: u64,
pub bytes: u64,
pub repo_finished: bool,
pub job_finished: bool,
pub complete: bool,
pub cancelled: bool,
pub message: String,
}
/// What a job looks like to a watcher, and to `Cancel`.
pub struct MigrationJob {
pub id: String,
/// Set by `Cancel`. The engine checks it at every job admission, so a cancel
/// takes effect within one in-flight slot's latency.
pub cancel: Arc<AtomicBool>,
/// The live feed. `broadcast` because a migration may have several watchers
/// (two operators, a UI and a terminal) and a watcher that drops must not
/// affect the job — with a channel that blocked on a full receiver, a
/// browser tab left open on a paused laptop would stall the migration.
tx: tokio::sync::broadcast::Sender<ProgressRow>,
/// The replay: the latest row per repository, plus the terminal row once
/// there is one. Guarded by the same lock the emitter takes, so a watcher
/// that subscribes under the lock can neither miss a row nor see one twice.
state: Mutex<JobState>,
}
#[derive(Default)]
struct JobState {
latest: Vec<ProgressRow>,
terminal: Option<ProgressRow>,
}
/// How many rows a slow watcher may fall behind before it starts losing them.
/// Progress rows are advisory — a watcher that lags gets the newer state, which
/// is the state it wants anyway — so lagging is survivable and blocking is not.
const PROGRESS_BACKLOG: usize = 1024;
impl MigrationJob {
fn new(id: String) -> Self {
let (tx, _rx) = tokio::sync::broadcast::channel(PROGRESS_BACKLOG);
Self {
id,
cancel: Arc::new(AtomicBool::new(false)),
tx,
state: Mutex::new(JobState::default()),
}
}
/// Record a row and publish it. Called from the engine's progress callback,
/// which is serial.
pub fn emit(&self, row: ProgressRow) {
{
let mut st = self.state.lock_or_recover();
if row.job_finished {
st.terminal = Some(row.clone());
} else {
match st.latest.iter_mut().find(|r| r.repo == row.repo) {
Some(slot) => *slot = row.clone(),
None => st.latest.push(row.clone()),
}
}
// Sent under the lock on purpose — see `subscribe`.
let _ = self.tx.send(row);
}
}
/// Join the feed. Returns the rows a late watcher missed (latest per
/// repository, then the terminal row if the job is already over) and a
/// receiver for everything after them.
///
/// Subscribing happens while holding the same lock `emit` takes, so there is
/// no window in which a row is neither in the replay nor in the stream, and
/// none in which it is in both.
pub fn subscribe(
&self,
) -> (
Vec<ProgressRow>,
tokio::sync::broadcast::Receiver<ProgressRow>,
) {
let st = self.state.lock_or_recover();
let rx = self.tx.subscribe();
let mut replay = st.latest.clone();
if let Some(t) = &st.terminal {
replay.push(t.clone());
}
(replay, rx)
}
/// True once the job has emitted its terminal row.
pub fn is_finished(&self) -> bool {
self.state.lock_or_recover().terminal.is_some()
}
}
/// Every migration this process has run, by id.
///
/// Deliberately in memory and deliberately bounded: a migration is a one-time
/// event an operator watches, not a record anyone queries a year later. What the
/// migration DID is in the audit log, which is the durable, tamper-evident place
/// for it; this is the live view while it runs.
pub struct MigrationJobs {
jobs: Mutex<HashMap<String, Arc<MigrationJob>>>,
seq: AtomicU64,
/// Finished jobs kept before the oldest is dropped.
keep: usize,
}
/// Finished jobs retained for inspection. Enough that an operator who steps away
/// can still find his run, small enough that a server migrating repeatedly cannot
/// grow this without bound.
const DEFAULT_JOBS_KEPT: usize = 32;
impl Default for MigrationJobs {
fn default() -> Self {
Self {
jobs: Mutex::new(HashMap::new()),
seq: AtomicU64::new(0),
keep: DEFAULT_JOBS_KEPT,
}
}
}
impl MigrationJobs {
/// Mint a job. The id carries the start time and a per-process counter: two
/// jobs started in the same millisecond still differ, and an operator reading
/// an id can see when it began without looking it up.
pub fn start(&self) -> Arc<MigrationJob> {
let n = self.seq.fetch_add(1, Ordering::Relaxed);
let millis = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis())
.unwrap_or(0);
let id = format!("mig-{millis}-{n}");
let job = Arc::new(MigrationJob::new(id.clone()));
let mut jobs = self.jobs.lock_or_recover();
jobs.insert(id, job.clone());
// Evict the oldest FINISHED jobs. A running job is never dropped, however
// old it is — a migration takes hours, and forgetting the one still going
// would leave it unwatchable and uncancellable.
if jobs.len() > self.keep {
let mut finished: Vec<String> = jobs
.iter()
.filter(|(_, j)| j.is_finished())
.map(|(k, _)| k.clone())
.collect();
finished.sort();
let excess = jobs.len().saturating_sub(self.keep);
for id in finished.into_iter().take(excess) {
jobs.remove(&id);
}
}
job
}
pub fn get(&self, id: &str) -> Option<Arc<MigrationJob>> {
self.jobs.lock_or_recover().get(id).cloned()
}
/// Ask a job to stop. `Ok(true)` when it was running and is now winding down,
/// `Ok(false)` when it had already finished — which is an answer, not an
/// error. `None` when there is no such job.
pub fn cancel(&self, id: &str) -> Option<bool> {
let job = self.get(id)?;
if job.is_finished() {
return Some(false);
}
job.cancel.store(true, Ordering::Relaxed);
Some(true)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Arc;
use traits::{ArtifactEntry, ArtifactFormat, ArtifactId, RepositoryBackendTrait};
/// A path-keyed in-memory repository behind the same HTTP door the real
/// backends implement — enough to prove the sink reads and writes through it.
struct PathRepo {
name: String,
writable: bool,
files: Mutex<HashMap<String, Vec<u8>>>,
}
impl PathRepo {
fn new(name: &str, writable: bool) -> Self {
Self {
name: name.into(),
writable,
files: Mutex::new(HashMap::new()),
}
}
fn relpath(&self, suburl: &str) -> Option<String> {
let mut segs = suburl.trim_start_matches('/').split('/');
match segs.next() {
Some(first) if first == self.name => {
let rest: Vec<&str> = segs.collect();
(!rest.is_empty()).then(|| rest.join("/"))
}
_ => None,
}
}
}
#[async_trait::async_trait]
impl RepositoryBackendTrait for PathRepo {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Raw
}
fn is_writable(&self) -> bool {
self.writable
}
fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("this test speaks the path door, not the coordinate door")
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
Ok(Vec::new())
}
fn handle_http2_request(
&self,
method: &str,
suburl: &str,
body: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
let Some(rel) = self.relpath(suburl) else {
return Ok((404, Vec::new(), b"no".to_vec()));
};
match method {
"GET" => match self.files.lock().unwrap().get(&rel) {
Some(b) => Ok((200, Vec::new(), b.clone())),
None => Ok((404, Vec::new(), b"no".to_vec())),
},
"PUT" => {
if !self.writable {
return Ok((403, Vec::new(), b"read-only".to_vec()));
}
self.files.lock().unwrap().insert(rel, body.to_vec());
Ok((201, Vec::new(), Vec::new()))
}
_ => Ok((405, Vec::new(), Vec::new())),
}
}
}
fn sink_with(repos: Vec<Arc<PathRepo>>) -> LocalRouteSink {
let routes: Vec<(String, Arc<dyn RepositoryBackendTrait>)> = repos
.into_iter()
.map(|r| (r.name.clone(), r as Arc<dyn RepositoryBackendTrait>))
.collect();
LocalRouteSink::new(FastRoutes::new(routes))
}
#[tokio::test]
async fn the_sink_stores_through_the_path_door_and_then_sees_it_is_present() {
let repo = Arc::new(PathRepo::new("crates", true));
let sink = sink_with(vec![repo.clone()]);
let body = b"crate-bytes";
let digest = nornir_hash::sha256_hex(body);
assert!(!sink
.already_has("crates", "a/a-1.0.crate", Some(&digest))
.await
.unwrap());
sink.store("crates", "a/a-1.0.crate", body).await.unwrap();
assert!(
sink.already_has("crates", "a/a-1.0.crate", Some(&digest))
.await
.unwrap(),
"RED-when-broken: this is the whole idempotence claim — re-running must see it"
);
assert_eq!(repo.files.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn a_stored_path_holding_different_bytes_is_not_the_same_artifact() {
let repo = Arc::new(PathRepo::new("crates", true));
let sink = sink_with(vec![repo]);
sink.store("crates", "a/a-1.0.crate", b"OLD").await.unwrap();
let other = nornir_hash::sha256_hex(b"NEW");
assert!(
!sink
.already_has("crates", "a/a-1.0.crate", Some(&other))
.await
.unwrap(),
"same path, different digest — the migration must converge, not skip"
);
}
#[tokio::test]
async fn a_repository_this_server_does_not_serve_holds_nothing_and_refuses_writes() {
let sink = sink_with(vec![]);
assert!(
!sink.already_has("absent", "x", None).await.unwrap(),
"absent is empty, not an error"
);
let err = sink
.store("absent", "x", b"y")
.await
.expect_err("cannot store into thin air");
assert!(err.to_string().contains("not served"), "{err}");
}
#[tokio::test]
async fn a_read_only_repository_refuses_before_the_backend_is_touched() {
let repo = Arc::new(PathRepo::new("sealed", false));
let sink = sink_with(vec![repo.clone()]);
let err = sink
.store("sealed", "x", b"y")
.await
.expect_err("read-only must refuse");
assert!(err.to_string().contains("read-only"), "{err}");
assert!(repo.files.lock().unwrap().is_empty());
}
#[tokio::test]
async fn the_server_sink_refuses_to_create_and_says_what_to_declare() {
let sink = sink_with(vec![]);
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}");
}
fn row(job: &str, repo: &str, done: u64) -> ProgressRow {
ProgressRow {
job_id: job.into(),
repo: repo.into(),
done,
failed: 0,
skipped: 0,
total: 10,
bytes: 0,
repo_finished: false,
job_finished: false,
complete: false,
cancelled: false,
message: String::new(),
}
}
#[tokio::test]
async fn a_watcher_that_joins_late_is_replayed_the_latest_row_per_repository() {
let jobs = MigrationJobs::default();
let job = jobs.start();
job.emit(row(&job.id, "a", 1));
job.emit(row(&job.id, "b", 1));
job.emit(row(&job.id, "a", 5)); // supersedes a=1
let (replay, _rx) = job.subscribe();
assert_eq!(replay.len(), 2, "one row per repository, not one per event");
let a = replay.iter().find(|r| r.repo == "a").unwrap();
assert_eq!(a.done, 5, "the LATEST row, not the first");
}
#[tokio::test]
async fn a_row_emitted_after_subscribing_arrives_on_the_stream() {
let jobs = MigrationJobs::default();
let job = jobs.start();
let (replay, mut rx) = job.subscribe();
assert!(replay.is_empty());
job.emit(row(&job.id, "a", 3));
let got = rx.recv().await.expect("the live feed carries it");
assert_eq!(got.done, 3);
// RED-when-broken: subscribe outside the emit lock and a row emitted in
// the window is in neither the replay nor the stream.
}
#[tokio::test]
async fn cancelling_a_running_job_sets_the_flag_and_cancelling_a_finished_one_is_not_an_error()
{
let jobs = MigrationJobs::default();
let job = jobs.start();
assert_eq!(jobs.cancel(&job.id), Some(true));
assert!(job.cancel.load(Ordering::Relaxed));
let mut terminal = row(&job.id, "a", 1);
terminal.job_finished = true;
job.emit(terminal);
assert_eq!(
jobs.cancel(&job.id),
Some(false),
"already over — an answer, not an error"
);
assert_eq!(
jobs.cancel("mig-nope"),
None,
"an unknown id is a different answer again"
);
}
#[tokio::test]
async fn job_ids_are_unique_within_the_same_millisecond() {
let jobs = MigrationJobs::default();
let a = jobs.start();
let b = jobs.start();
assert_ne!(a.id, b.id);
assert!(jobs.get(&a.id).is_some());
assert!(jobs.get(&b.id).is_some());
}
#[tokio::test]
async fn a_running_job_is_never_evicted_however_many_follow_it() {
let jobs = MigrationJobs {
keep: 2,
..Default::default()
};
let long_runner = jobs.start();
for _ in 0..10 {
let j = jobs.start();
let mut t = row(&j.id, "x", 0);
t.job_finished = true;
j.emit(t);
}
assert!(
jobs.get(&long_runner.id).is_some(),
"RED-when-broken: evict by age alone and the migration still running becomes \
unwatchable and uncancellable"
);
}
}