server/lib/tests/replication_runs_as_a_resumable_job.rs
//! ★ **Replication runs as a resumable job, through the real doors** (lane OX-WIRED4).
//!
//! A running holger (`mirror`: the ReplicationService door and the job queue) with two `raw` repositories (`dev`, `prod`), static-token principals
//! (`rel` Admin, `rita` Reader) and its job queue under the data root, asked by real tonic
//! clients:
//!
//! * `ReplicationService.Replicate` with `execute` answers a JOB id and zero counts (nothing
//! has run yet); `JobService.GetJob` follows it to `done`, its result names the counts,
//! and `prod` serves the exact bytes over `RepositoryService.FetchArtifact`;
//! * a second run over the unchanged source is a job that copies nothing (content-id
//! idempotent); a Reader is refused and admits no job;
//! * a replication ADMITTED by a process that died before it ran (its row left `queued` in
//! `<data root>/jobs`) is resumed by the next `Holger::start` and runs to `done`: the
//! boot's own recovery, the boot's own runner, read back over the job door.
//!
//! Before, the queued shape was tested only at the `HolgerGrpc` level (services.rs unit
//! tests: the job is admitted and stays `queued`; no runner, no booted server) and the
//! recovery only on a `JobQueue` in a tempdir (jobs.rs).
use std::io::Write;
use std::time::{Duration, Instant};
use holger_server_lib::grpc::holger_proto::job_service_client::JobServiceClient;
use holger_server_lib::grpc::holger_proto::replication_service_client::ReplicationServiceClient;
use holger_server_lib::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use holger_server_lib::grpc::holger_proto::{
ArtifactId as ProtoId, FetchArtifactRequest, GetJobRequest, JobRow, ListJobsRequest,
ReplicateRequest,
};
use holger_server_lib::jobs::{status, Job, JobQueue};
use holger_server_lib::{read_ron_config, wire_holger, ArtifactId, Holger};
use traits::test_support::hold_port;
const ADMIN: &str = "rel-admin-token-of-sufficient-length-for-holger";
const READER: &str = "rita-token-of-sufficient-length-for-holger";
const FILES: [(&str, &[u8]); 2] = [
("tools/a.bin", b"the bytes of a"),
("tools/b.bin", b"the bytes of b, a little longer"),
];
struct Running {
_data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
_holger: Holger,
grpc: String,
}
fn ron(d: &str, grpc: u16, http: u16) -> String {
let repo = |name: &str| {
format!(
r#"(ron_name: "{name}", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")),
ron_data_dir: Some("{d}/{name}")),"#
)
};
format!(
r#"(
repositories: [ {dev} {prod} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
auth: Some((
methods: [ StaticToken(subject: "rel", token: Some("{ADMIN}")), StaticToken(subject: "rita", token: Some("{READER}")) ],
roles: {{"rel": Admin, "rita": Reader}},
)),
)"#,
dev = repo("dev"),
prod = repo("prod"),
)
}
fn id(name: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.into(),
version: String::new(),
}
}
/// Boot a holger whose `dev` holds [`FILES`]; `before_start` runs on the data root
/// after the seed and before `Holger::start` (the leftovers of a process that died).
fn start(before_start: impl Fn(&std::path::Path)) -> Running {
let mut last = String::new();
for _ in 0..8 {
let (grpc, http) = (hold_port(), hold_port());
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_string();
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron(&d, grpc, http).as_bytes()).unwrap();
let go = || -> Result<Holger, String> {
let mut h = read_ron_config(cfg.path()).map_err(|e| format!("config: {e:#}"))?;
h.instantiate_backends()
.map_err(|e| format!("backends: {e:#}"))?;
for (name, bytes) in FILES {
h.put("dev", &id(name), bytes)
.map_err(|e| format!("seed: {e:#}"))?;
}
wire_holger(&mut h).map_err(|e| format!("wire: {e:#}"))?;
assert_eq!(
h.data_root(),
data.path(),
"the job queue lives under the data root"
);
before_start(data.path());
h.start().map_err(|e| format!("start: {e:#}"))?;
Ok(h)
};
match go() {
Ok(h) => {
return Running {
_data: data,
_cfg: cfg,
_holger: h,
grpc: format!("http://127.0.0.1:{grpc}"),
}
}
Err(e) if e.starts_with("config") => panic!("{e}"),
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
fn as_<T>(token: &str, body: T) -> tonic::Request<T> {
let mut r = tonic::Request::new(body);
r.metadata_mut()
.insert("authorization", format!("Bearer {token}").parse().unwrap());
r
}
type Clients = (
ReplicationServiceClient<tonic::transport::Channel>,
JobServiceClient<tonic::transport::Channel>,
RepositoryServiceClient<tonic::transport::Channel>,
);
async fn clients(r: &Running) -> Clients {
for _ in 0..250 {
if let Ok(ch) = tonic::transport::Endpoint::from_shared(r.grpc.clone())
.unwrap()
.connect()
.await
{
return (
ReplicationServiceClient::new(ch.clone()),
JobServiceClient::new(ch.clone()),
RepositoryServiceClient::new(ch),
);
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("holger never answered at {}", r.grpc);
}
fn replicate() -> ReplicateRequest {
ReplicateRequest {
from_repo: "dev".into(),
to_repo: "prod".into(),
execute: true,
limit: 0,
async_job: false,
}
}
/// Follow a job over `JobService.GetJob` until it is terminal.
async fn finished(jobs: &mut JobServiceClient<tonic::transport::Channel>, job_id: &str) -> JobRow {
let deadline = Instant::now() + Duration::from_secs(30);
loop {
let got = jobs
.get_job(as_(
ADMIN,
GetJobRequest {
job_id: job_id.into(),
},
))
.await
.expect("the job door answers")
.into_inner();
assert!(got.found, "the ledger holds job {job_id}");
let row = got.job.unwrap();
if row.status != status::QUEUED && row.status != status::RUNNING {
return row;
}
assert!(
Instant::now() < deadline,
"job {job_id} never finished: {row:?}"
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
async fn fetch(
repo: &mut RepositoryServiceClient<tonic::transport::Channel>,
name: &str,
) -> Option<Vec<u8>> {
let req = FetchArtifactRequest {
repository: "prod".into(),
id: Some(ProtoId {
namespace: String::new(),
name: name.into(),
version: String::new(),
}),
};
match repo.fetch_artifact(as_(ADMIN, req)).await {
Ok(r) => Some(r.into_inner().data.to_vec()),
Err(st) if st.code() == tonic::Code::NotFound => None,
Err(st) => panic!("fetch {name}: {st:?}"),
}
}
#[tokio::test(flavor = "multi_thread")]
async fn an_executing_replication_is_a_job_the_booted_server_runs_to_done() {
let r = start(|_| {});
let (mut repl, mut jobs, mut repo) = clients(&r).await;
// A Reader is refused, and no job is admitted.
let st = repl
.replicate(as_(READER, replicate()))
.await
.expect_err("a Reader does not replicate");
assert!(
matches!(
st.code(),
tonic::Code::PermissionDenied | tonic::Code::Unauthenticated
),
"{st:?}"
);
let listed = jobs
.list_jobs(as_(
ADMIN,
ListJobsRequest {
kind: String::new(),
limit: 0,
..Default::default()
},
))
.await
.unwrap()
.into_inner();
assert!(
listed.jobs.iter().all(|j| j.kind != "holger_replicate"),
"the refusal admitted nothing: {:?}",
listed.jobs
);
// The admin's executing replication is answered with a job id and no counts.
let reply = repl
.replicate(as_(ADMIN, replicate()))
.await
.expect("admitted")
.into_inner();
assert!(
!reply.job_id.is_empty(),
"an executing replication on a server with a queue is a job: {reply:?}"
);
assert_eq!(
(reply.copied, reply.examined),
(0, 0),
"nothing has run when the answer is given"
);
let row = finished(&mut jobs, &reply.job_id).await;
assert_eq!(row.status, status::DONE, "{row:?}");
assert_eq!(row.kind, "holger_replicate", "{row:?}");
assert_eq!(row.requested_by, "rel", "{row:?}");
assert!(
row.result_ref.contains("copied=2"),
"the result names the counts: {row:?}"
);
for (name, bytes) in FILES {
assert_eq!(
fetch(&mut repo, name).await.as_deref(),
Some(bytes),
"prod serves {name}"
);
}
// Again over the unchanged source: a job that copies nothing.
let again = repl
.replicate(as_(ADMIN, replicate()))
.await
.expect("admitted")
.into_inner();
assert_ne!(
again.job_id, reply.job_id,
"a finished job is not the open work of a new run"
);
let row = finished(&mut jobs, &again.job_id).await;
assert_eq!(row.status, status::DONE, "{row:?}");
assert!(
row.result_ref.contains("copied=0") && row.result_ref.contains("skipped=2"),
"idempotent by content id: {row:?}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn a_replication_admitted_before_the_process_died_resumes_at_the_next_start() {
let left = std::sync::Mutex::new(String::new());
let r = start(|root| {
// The process that died: it admitted the replication (the ledger row is `queued`)
// and never ran it (no runner was started).
let (queue, _) = JobQueue::open(root, 0).expect("the queue under the data root");
let job = Job::Replicate {
from_repo: "dev".into(),
to_repo: "prod".into(),
execute: true,
limit: 0,
requested_by: "rel".into(),
};
let job_id = queue.enqueue(job).expect("admitted");
assert_eq!(queue.get(&job_id).unwrap().unwrap().status, status::QUEUED);
*left.lock().unwrap() = job_id.0.clone();
});
let job_id = left.lock().unwrap().clone();
let (_, mut jobs, mut repo) = clients(&r).await;
let row = finished(&mut jobs, &job_id).await;
assert_eq!(
row.status,
status::DONE,
"the boot resumed the job and ran it: {row:?}"
);
assert!(row.result_ref.contains("copied=2"), "{row:?}");
for (name, bytes) in FILES {
assert_eq!(
fetch(&mut repo, name).await.as_deref(),
Some(bytes),
"prod serves {name} after the resumed run"
);
}
}