server/lib/tests/a_sold_repository_serves_at_once.rs
//! ★ **A repository a paid signup made serves at once** (lane M4b). The ONE monetize's
//! pulled `add_principal` (lane M4's path) makes the buyer's repository; with no restart, an
//! upload and a download reach it over the box's own HTTP door, the gRPC door resolves it,
//! the `repositories_wanted` health word is clear, and its paid limit measures it. It is
//! private (lane P1): a stranger is asked for a credential, the estate's admin reads it. Requests
//! running while the route table is swapped stay correct: downloads from the estate's own
//! repository answer the same bytes throughout.
use std::io::Write as _;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use traits::test_support::hold_port;
use ed25519_dalek::{Signer, SigningKey};
use holger_fleet::exec::{execute, Ctx, NoAsk};
use holger_fleet::{Addressee, HolgerBox, Order};
use holger_server_lib::fleet_box::{FleetBox, REPOSITORIES_WANTED};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use monetize_product::pull::AddPrincipal;
use monetize_product::{EntitlementFact, State, TenantId};
const TOKEN: &str = "c0nsole-bearer-0123456789abcdefghij";
fn fact(key: &SigningKey, repo: &str, cap: u64) -> EntitlementFact {
let mut f = EntitlementFact {
tenant: TenantId(repo.into()),
plan: format!("holger/{repo}/2026-10-19+disk_bytes"),
state: State::Paid,
paid_until_unix_ms: Some(4_000_000_000_000),
caps: [("disk_bytes".to_string(), cap)].into_iter().collect(),
source: "payment:stripe:o-1".into(),
signature: Vec::new(),
issued_unix_ms: Some(1),
issued_signature: Vec::new(),
};
f.signature = key
.sign(&monetize_embed::signing::fact_message(&f))
.to_bytes()
.to_vec();
f.issued_signature = key
.sign(&monetize_embed::signing::fact_message_issued(&f))
.to_bytes()
.to_vec();
f
}
fn config(d: &std::path::Path, key: &SigningKey, grpc: u16, http: u16) -> tempfile::NamedTempFile {
let ron = format!(
r#"(
repositories: [
(ron_name: "crates", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{d}/crates")),
(ron_name: "monetize-backup", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{d}/monetize-backup"), ron_require_auth: true, ron_require_role: Some(Admin)),
],
exposed_endpoints: [
(ron_name: "default", 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 ),
access: (reads: Open, front: Open), auth: Some(( methods: [ StaticToken(subject: "console", token: Some("{TOKEN}")) ], roles: {{ "console": Admin }}, require_auth_for_reads: false )),
monetize_pubkey: Some("{pk}"),
)"#,
d = d.display(),
pk = hex::encode(key.verifying_key().to_bytes()),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
cfg
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_repository_made_by_add_principal_takes_an_upload_and_serves_it_with_no_restart() {
let data = tempfile::tempdir().unwrap();
let d = data.path();
std::fs::create_dir_all(d.join("audit")).unwrap();
let key = SigningKey::from_bytes(&[21u8; 32]);
let http = hold_port();
let cfg = config(d, &key, hold_port(), http);
let mut holger: Holger = read_ron_config(cfg.path()).unwrap();
holger.instantiate_backends().unwrap();
wire_holger(&mut holger).unwrap();
holger.start().unwrap();
let (state, own) =
holger_server_lib::fleet_box::served_state().expect("the endpoint published its state");
let cl = reqwest::Client::new();
let base = format!("http://127.0.0.1:{http}");
for _ in 0..300 {
if cl
.get(format!("{base}/healthz"))
.send()
.await
.map(|r| r.status() == 200)
.unwrap_or(false)
{
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
let put = |repo: &str, name: &str, body: Vec<u8>| {
cl.put(format!("{base}/{repo}/{name}"))
.bearer_auth(TOKEN)
.body(body)
.send()
};
// Before: the name is no repository here.
assert_eq!(
cl.get(format!("{base}/acme-artifacts/x"))
.send()
.await
.unwrap()
.status()
.as_u16(),
404
);
assert!(put("crates", "own.tar", b"the estate's own".to_vec())
.await
.unwrap()
.status()
.is_success());
// Downloads from the estate's own repository run WHILE the table is swapped.
let stop = Arc::new(AtomicBool::new(false));
let hammer = {
let (cl, base, stop) = (cl.clone(), base.clone(), stop.clone());
tokio::spawn(async move {
let mut n = 0u32;
while !stop.load(Ordering::Relaxed) || n == 0 {
let r = cl
.get(format!("{base}/crates/own.tar"))
.send()
.await
.unwrap();
assert_eq!(
r.status().as_u16(),
200,
"a download in flight across the swap stays served"
);
assert_eq!(r.bytes().await.unwrap().as_ref(), b"the estate's own");
n += 1;
}
n
})
};
let b = FleetBox {
grpc: state.clone(),
own,
rt: tokio::runtime::Handle::current(),
bearer: None.into(),
build: "test".into(),
recharge_failed: Default::default(),
};
let me = Addressee {
product: "holger".into(),
instance: "acme".into(),
role: "main".into(),
};
let vk = monetize_embed::VerifyingKey::from_bytes(&key.verifying_key().to_bytes()).unwrap();
let key2 = key.clone();
let (acks, health) = tokio::task::spawn_blocking(move || {
let ctx = Ctx {
me: &me,
key: &vk,
now_unix_ms: holger_fleet::now_unix_ms(),
};
let signup = Order::AddPrincipal(AddPrincipal {
username: "acme-artifacts".into(),
kind: "org".into(),
signup: "s-1".into(),
});
let a = execute(&b, &ctx, 1, &signup, &mut NoAsk);
let f = execute(
&b,
&ctx,
2,
&Order::Entitlement(fact(&key2, "acme-artifacts", 100_000)),
&mut NoAsk,
);
let again = execute(&b, &ctx, 3, &signup, &mut NoAsk);
(vec![a, f, again], b.health())
})
.await
.unwrap();
assert!(
acks.iter().all(|a| a.ok),
"made, entitled, asked again: {acks:?}"
);
assert!(
!health.contains_key(REPOSITORIES_WANTED),
"served, so nothing is wanted: {health:?}"
);
// No restart: the upload lands and the download answers its bytes.
let up = put("acme-artifacts", "first.tar", b"the buyer's first".to_vec())
.await
.unwrap();
assert!(
up.status().is_success(),
"an upload reaches the new repository at once: {}",
up.status()
);
// ★ Lane P1: private, never its template's open read: a stranger is asked for a
// credential, the estate's admin reads it.
let anonymous = cl
.get(format!("{base}/acme-artifacts/first.tar"))
.send()
.await
.unwrap();
assert_eq!(
anonymous.status().as_u16(),
401,
"a sold repository never copies its template's world read"
);
let down = cl
.get(format!("{base}/acme-artifacts/first.tar"))
.bearer_auth(TOKEN)
.send()
.await
.unwrap();
assert_eq!(down.status().as_u16(), 200);
assert_eq!(down.bytes().await.unwrap().as_ref(), b"the buyer's first");
assert!(
d.join("acme-artifacts").exists(),
"its own bytes, beside the template's: {}",
d.display()
);
assert!(
state.routes.lookup("acme-artifacts").is_some(),
"the gRPC door resolves it too"
);
// Its paid limit measures it where its own bytes are.
let big = put("acme-artifacts", "big.tar", vec![7u8; 200_000])
.await
.unwrap();
assert_eq!(big.status().as_u16(), 507, "past its paid limit is refused");
let why = big.text().await.unwrap();
let place = d.join("acme-artifacts").display().to_string();
assert!(
why.contains("`acme-artifacts` is full") && !why.contains(&place),
"its own room, and no disk path shown to the customer (A10 row 18): {why}"
);
stop.store(true, Ordering::Relaxed);
let served = hammer.await.unwrap();
assert!(served > 0, "the downloads ran across the swap");
}