server/lib/tests/sync_between_two_holgers.rs
//! **The sync protocol between two Holgers** (Holger v2.0 H4, `server_lib::sync`).
//!
//! Two real Holgers run in this process, each with its own intake chain, HTTP door,
//! audit log and sync key: the upstream `dmz` and the downstream `onprem`.
//!
//! * USB mode: the upstream invalidates a package; its signed manifest goes onto a
//! directory that stands in for the stick; the downstream verifies it under its
//! trust list, holds the package at once (its PEP 503 page drops it, the file is
//! refused) and writes its signed ack file back onto the stick; the upstream
//! imports the ack and the alarm it raised for the missing ack clears.
//! * Pull mode: the downstream fetches the manifest from the upstream's door,
//! applies it and posts its ack back; the upstream's board shows the downstream
//! current and the invalidation reached.
//! * The alarm: no ack within the deadline raises one alarm per downstream and
//! invalidation (audit, board flag, the operator's log line), once.
// A server's test, never a gateway build's (lane HOL-CLEANUP): a gateway build is a
// downstream, and an upstream and a downstream in one process, and a downstream's egress pin is process-wide.
#![cfg(not(feature = "gateway"))]
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use holger_server_lib::promote::intake::{ArtifactState, IntakeChain, TransitionRequest};
use holger_server_lib::sync::{HttpLink, SyncHub};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use holger_sync::{Reason, SyncKey};
use traits::test_support::hold_port;
use traits::ArtifactId;
const WHEEL_A: &str = "alpha_pkg-1.0.0-py3-none-any.whl";
const WHEEL_B: &str = "beta_pkg-2.0.0-py3-none-any.whl";
fn token(who: &str) -> String {
format!("{who}-token-of-sufficient-length-for-holger")
}
/// The HTTP road, recording every call's method, url and bearer: what a real downstream sends.
struct Recording {
inner: HttpLink,
seen: std::sync::Mutex<Vec<(String, String, Option<String>)>>,
}
impl Recording {
fn new() -> Self {
Self {
inner: HttpLink::new(),
seen: Default::default(),
}
}
}
impl holger_server_lib::sync::SyncLink for Recording {
fn get(&self, url: &str, token: Option<&str>) -> Result<Vec<u8>, String> {
self.seen
.lock()
.unwrap()
.push(("GET".into(), url.into(), token.map(String::from)));
self.inner.get(url, token)
}
fn post(&self, url: &str, body: &[u8], token: Option<&str>) -> Result<Vec<u8>, String> {
self.seen
.lock()
.unwrap()
.push(("POST".into(), url.into(), token.map(String::from)));
self.inner.post(url, body, token)
}
}
fn crc32(data: &[u8]) -> u32 {
let mut crc = 0xFFFF_FFFFu32;
for b in data {
crc ^= *b as u32;
for _ in 0..8 {
crc = if crc & 1 != 0 {
(crc >> 1) ^ 0xEDB8_8320
} else {
crc >> 1
};
}
}
!crc
}
/// A small, well-formed wheel: a stored-only zip with its dist-info.
fn wheel(project: &str, version: &str) -> Vec<u8> {
let module = project.replace('-', "_");
let di = format!("{module}-{version}.dist-info");
let members: Vec<(String, Vec<u8>)> = vec![
(format!("{module}/__init__.py"), format!("__version__ = \"{version}\"\n").into_bytes()),
(format!("{di}/METADATA"), format!("Metadata-Version: 2.1\nName: {project}\nVersion: {version}\nLicense-Expression: MIT\n").into_bytes()),
(format!("{di}/WHEEL"), b"Wheel-Version: 1.0\nGenerator: holger-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n".to_vec()),
];
let mut out = Vec::new();
let mut central = Vec::new();
for (name, body) in &members {
let crc = crc32(body);
let offset = out.len() as u32;
out.extend_from_slice(&0x0403_4b50u32.to_le_bytes());
out.extend_from_slice(&[20, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
out.extend_from_slice(&crc.to_le_bytes());
out.extend_from_slice(&(body.len() as u32).to_le_bytes());
out.extend_from_slice(&(body.len() as u32).to_le_bytes());
out.extend_from_slice(&(name.len() as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out.extend_from_slice(name.as_bytes());
out.extend_from_slice(body);
central.extend_from_slice(&0x0201_4b50u32.to_le_bytes());
central.extend_from_slice(&[20, 0, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
central.extend_from_slice(&crc.to_le_bytes());
central.extend_from_slice(&(body.len() as u32).to_le_bytes());
central.extend_from_slice(&(body.len() as u32).to_le_bytes());
central.extend_from_slice(&(name.len() as u16).to_le_bytes());
central.extend_from_slice(&[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
central.extend_from_slice(&offset.to_le_bytes());
central.extend_from_slice(name.as_bytes());
}
let cd_offset = out.len() as u32;
let n = members.len() as u16;
out.extend_from_slice(¢ral);
out.extend_from_slice(&0x0605_4b50u32.to_le_bytes());
out.extend_from_slice(&[0, 0, 0, 0]);
out.extend_from_slice(&n.to_le_bytes());
out.extend_from_slice(&n.to_le_bytes());
out.extend_from_slice(&(central.len() as u32).to_le_bytes());
out.extend_from_slice(&cd_offset.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes());
out
}
fn id(name: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.into(),
version: String::new(),
}
}
// ── one Holger ──────────────────────────────────────────────────────────────
struct Running {
holger: Holger,
_rt: tokio::runtime::Runtime,
data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
http: String,
}
impl Running {
fn chain(&self) -> Arc<IntakeChain> {
self.holger.intake().expect("the intake chain")
}
fn hub(&self) -> Arc<SyncHub> {
self.chain().sync().expect("the sync hub")
}
fn audit_text(&self) -> String {
std::fs::read_dir(self.data.path().join("audit"))
.unwrap()
.flatten()
.map(|e| std::fs::read_to_string(e.path()).unwrap_or_default())
.collect()
}
}
fn start(sync: &str) -> Running {
let mut last = String::new();
for _ in 0..8 {
match try_start(hold_port(), hold_port(), sync) {
Ok(r) => return r,
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
fn try_start(grpc: u16, http: u16, sync: &str) -> Result<Running, String> {
let data = tempfile::tempdir().unwrap();
let d = data.path().display().to_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}")),"#
)
};
let methods: Vec<String> = ["ida", "bob", "aud", "onprem"]
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
let ron = format!(
r#"(
// Both holgers share this test process, and the egress guard is per process: the
// upstream names no upstream of its own, so it says `network: Online` (B2).
network: Online,
repositories: [ {incoming} {quarantine} {approved} {production} ],
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: [ {methods} ],
duties: {{ "ida": [Importer], "bob": [SecurityApprover], "aud": [Auditor] }},
)),
promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"), (name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"), (name: "production", repo: "production"),
],
intake: Some((
incoming: "incoming", quarantine: "quarantine", approved: "approved", production: "production",
rules: (sandbox: off),
approvals: 1,
scan: None,
)),
)),
sync: Some(({sync})),
)"#,
incoming = repo("incoming"),
quarantine = repo("quarantine"),
approved = repo("approved"),
// HCON-5: an estate build gates reads; the repository clients read is public (as the
// ONE box feeds an estate's `public_repositories`), the intake stages stay gated.
production =
repo("production").replacen("ron_in: None,", "ron_in: None, visibility: Public,", 1),
methods = methods.join(", "),
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron.as_bytes()).unwrap();
let mut holger: Holger = read_ron_config(cfg.path()).map_err(|e| format!("config: {e}"))?;
holger
.instantiate_backends()
.map_err(|e| format!("backends: {e}"))?;
wire_holger(&mut holger).map_err(|e| format!("wire: {e:#}"))?;
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.enable_all()
.build()
.unwrap();
{
let _in = rt.enter();
holger.start().map_err(|e| format!("start: {e:#}"))?;
}
let r = Running {
holger,
_rt: rt,
data,
_cfg: cfg,
http: format!("127.0.0.1:{http}"),
};
for _ in 0..200 {
if request(&r.http, "GET", "/healthz", None, &[]).status == 200 {
return Ok(r);
}
std::thread::sleep(Duration::from_millis(20));
}
Err("never healthy".into())
}
struct Got {
status: u16,
body: Vec<u8>,
}
impl Got {
fn text(&self) -> String {
String::from_utf8_lossy(&self.body).to_string()
}
fn json(&self) -> serde_json::Value {
serde_json::from_slice(&self.body).unwrap_or_else(|e| panic!("{e}: {}", self.text()))
}
}
fn request(addr: &str, method: &str, path: &str, bearer: Option<&str>, body: &[u8]) -> Got {
let Ok(mut s) = TcpStream::connect(addr) else {
return Got {
status: 0,
body: Vec::new(),
};
};
let mut req = format!(
"{method} {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\nContent-Length: {}\r\n",
body.len()
);
if let Some(t) = bearer {
req.push_str(&format!("Authorization: Bearer {t}\r\n"));
}
req.push_str("\r\n");
s.write_all(req.as_bytes()).unwrap();
s.write_all(body).unwrap();
let mut all = Vec::new();
let _ = s.read_to_end(&mut all);
let split = all
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|i| i + 4)
.unwrap_or(all.len());
let head = String::from_utf8_lossy(&all[..split]).to_string();
let status = head
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(0);
Got {
status,
body: all[split..].to_vec(),
}
}
/// Upload `name` as the importer and walk it to AVAILABLE as the approver.
fn land(r: &Running, name: &str, bytes: &[u8]) {
let put = request(
&r.http,
"PUT",
&format!("/incoming/{name}"),
Some(&token("ida")),
bytes,
);
assert_eq!(put.status, 201, "{}", put.text());
let chain = r.chain();
for to in [
ArtifactState::Scanned,
ArtifactState::PolicyChecked,
ArtifactState::Approved,
ArtifactState::Sealed,
ArtifactState::Promoted,
ArtifactState::Available,
] {
chain
.advance(TransitionRequest::new(id(name), to, "bob"))
.unwrap_or_else(|e| panic!("{name} → {to}: {e}"));
}
}
fn simple_page(r: &Running, project: &str) -> Got {
request(
&r.http,
"GET",
&format!("/production/simple/{project}/"),
None,
&[],
)
}
/// The two keys, minted before the configs that name each other's.
struct Keys {
_dir: tempfile::TempDir,
up: PathBuf,
down: PathBuf,
up_hex: String,
down_hex: String,
}
fn keys() -> Keys {
let dir = tempfile::tempdir().unwrap();
let (up, down) = (dir.path().join("dmz.key"), dir.path().join("onprem.key"));
let up_hex = SyncKey::open_or_mint(&up).unwrap().public_hex();
let down_hex = SyncKey::open_or_mint(&down).unwrap().public_hex();
Keys {
_dir: dir,
up,
down,
up_hex,
down_hex,
}
}
fn upstream_ron(k: &Keys, deadline: u64) -> String {
format!(
r#"name: "dmz", key: Some("{}"), downstreams: [ (name: "onprem", key: "{}") ], ack_deadline_secs: {deadline}"#,
k.up.display(),
k.down_hex
)
}
fn downstream_ron(k: &Keys, link: &str) -> String {
format!(
r#"name: "onprem", key: Some("{}"), upstreams: [ (name: "dmz", key: "{}"{link}) ]"#,
k.down.display(),
k.up_hex
)
}
fn now() -> i64 {
holger_sync::now_secs()
}
fn read_dir_names(p: &Path) -> Vec<String> {
let mut v: Vec<String> = std::fs::read_dir(p)
.unwrap()
.flatten()
.map(|e| e.file_name().to_string_lossy().to_string())
.collect();
v.sort();
v
}
// ── USB mode ────────────────────────────────────────────────────────────────
#[test]
fn usb_an_invalidation_is_blocked_downstream_and_its_ack_clears_the_alarm() {
let k = keys();
let up = start(&upstream_ron(&k, 60));
let down = start(&downstream_ron(&k, ""));
let (a, b) = (wheel("alpha-pkg", "1.0.0"), wheel("beta-pkg", "2.0.0"));
for r in [&up, &down] {
land(r, WHEEL_A, &a);
land(r, WHEEL_B, &b);
}
let stick = tempfile::tempdir().unwrap();
let (hu, hd) = (up.hub(), down.hub());
assert_eq!(hu.public_key(), k.up_hex);
// 1. Generation 1: the packages, no invalidation. Stick → downstream → ack → stick → upstream.
let m1 = hu.usb_manifest(0).unwrap();
let m1_path = hu.write_manifest(&m1, stick.path()).unwrap();
let applied = hd
.apply(&std::fs::read(&m1_path).unwrap(), "usb", "operator", "usb")
.unwrap();
assert_eq!((applied.generation, applied.added), (1, 2));
assert!(applied.quarantined.is_empty());
let ack1 = hd.write_ack(&applied.ack, stick.path()).unwrap();
let out = hu
.import_ack(&std::fs::read(&ack1).unwrap(), "usb", "usb")
.unwrap();
assert_eq!(out.generation, 1);
let board = hu.board(now()).unwrap();
assert_eq!(board.downstreams[0].state, "current", "{board:?}");
// 2. The upstream invalidates A: a new generation, and A is held on the upstream too.
let inv = hu
.invalidate(
"sec-anna",
"local",
WHEEL_A,
None,
Reason::Malware,
"a backdoor found after its approval",
)
.unwrap();
assert!(inv.new);
assert_eq!(
inv.invalidation.sha256.as_deref(),
Some(update_bundle::digest::sha256_hex(&a).as_str()),
"the bytes are named"
);
assert_eq!(up.chain().state(&id(WHEEL_A)), ArtifactState::Quarantined);
assert_eq!(
simple_page(&up, "alpha-pkg").status,
404,
"the upstream no longer serves it either"
);
// 3. No ack within the deadline: ONE alarm, on the board, in the audit and in the operator's line.
let t = inv.invalidation.issued_at;
assert!(
hu.check_alarms(t + 30).unwrap().is_empty(),
"inside the deadline"
);
let alarms = hu.check_alarms(t + 61).unwrap();
assert_eq!(alarms.len(), 1, "{alarms:?}");
assert_eq!(alarms[0].downstream, "onprem");
assert!(hu.check_alarms(t + 62).unwrap().is_empty(), "raised once");
let board = hu.board(t + 62).unwrap();
assert!(board.alarm);
assert_eq!(board.downstreams[0].state, "alarm");
assert_eq!(board.invalidations[0].pending, vec!["onprem".to_string()]);
assert!(
up.audit_text()
.contains("ALARM sync: downstream 'onprem' has not acknowledged invalidation"),
"the alarm is audited"
);
// 4. The stick carries the new manifest; the downstream holds A at once and every view drops it.
let m2 = hu.usb_manifest(1).unwrap();
let m2_path = hu.write_manifest(&m2, stick.path()).unwrap();
assert_eq!(
simple_page(&down, "alpha-pkg").status,
200,
"served before the sync"
);
let applied = hd
.apply(&std::fs::read(&m2_path).unwrap(), "usb", "operator", "usb")
.unwrap();
assert_eq!(applied.generation, 2);
assert_eq!(applied.quarantined, vec![WHEEL_A.to_string()]);
assert_eq!(down.chain().state(&id(WHEEL_A)), ArtifactState::Quarantined);
assert_eq!(
simple_page(&down, "alpha-pkg").status,
404,
"the PEP 503 page drops it"
);
assert_eq!(simple_page(&down, "beta-pkg").status, 200, "B is untouched");
assert_eq!(
request(
&down.http,
"GET",
&format!("/production/{WHEEL_A}"),
None,
&[]
)
.status,
404,
"the file is refused"
);
let last = down.chain().history(&id(WHEEL_A)).last().cloned().unwrap();
assert!(
last.reason
.starts_with("invalidated by upstream 'dmz' (generation 2): malware"),
"{}",
last.reason
);
assert!(
down.audit_text().contains("\"invalidated\""),
"the downstream's trail carries it"
);
// 5. The ack file goes back on the stick; the upstream imports it and the alarm clears.
let ack2 = hd.write_ack(&applied.ack, stick.path()).unwrap();
assert_eq!(
ack2.file_name().unwrap().to_string_lossy(),
"holger-sync-ack-onprem-2.json"
);
assert!(
read_dir_names(stick.path()).contains(&"holger-sync-ack-onprem-2.json".to_string()),
"{:?}",
read_dir_names(stick.path())
);
let out = hu
.import_ack(&std::fs::read(&ack2).unwrap(), "usb", "usb")
.unwrap();
assert_eq!(out.cleared, vec![inv.invalidation.id.clone()]);
let board = hu.board(t + 63).unwrap();
assert!(!board.alarm, "{board:?}");
assert_eq!(board.downstreams[0].state, "current");
assert_eq!(board.downstreams[0].invalidations_reached, 1);
assert_eq!(board.invalidations[0].reached, vec!["onprem".to_string()]);
assert!(
up.audit_text().contains("alarm cleared"),
"the clearing is audited"
);
// 6. A replayed old stick is a rollback; a manifest signed by another key is refused.
let e = hd
.apply(&std::fs::read(&m1_path).unwrap(), "usb", "operator", "usb")
.unwrap_err();
assert!(e.tagged().contains("[holger-sync-4]"), "{}", e.tagged());
let forged = {
let mut m = holger_sync::open_manifest(
&std::fs::read(&m2_path).unwrap(),
&[holger_sync::TrustedPeer {
name: "dmz".into(),
key: k.up_hex.clone(),
}],
)
.unwrap();
m.invalidations.clear();
m.generation = 3;
SyncKey::open_or_mint(&k.down)
.unwrap()
.sign_manifest(&m)
.unwrap()
};
let e = hd
.apply(&forged.to_bytes(), "usb", "operator", "usb")
.unwrap_err();
assert!(e.tagged().contains("[holger-sync-2]"), "{}", e.tagged());
// An ack the downstream did not sign is refused at the upstream.
let mut wrong = applied.ack.clone();
wrong.signer = "elsewhere".into();
let e = hu.import_ack(&wrong.to_bytes(), "usb", "usb").unwrap_err();
assert!(e.tagged().contains("[holger-sync-5]"), "{}", e.tagged());
}
// ── pull mode ───────────────────────────────────────────────────────────────
#[test]
fn pull_the_downstream_fetches_applies_and_acknowledges_over_the_door() {
let k = keys();
let up = start(&upstream_ron(&k, 3600));
let tok = tempfile::NamedTempFile::new().unwrap();
std::fs::write(tok.path(), token("onprem")).unwrap();
let down = start(&downstream_ron(
&k,
&format!(
r#", url: Some("http://{}"), token_file: Some("{}")"#,
up.http,
tok.path().display()
),
));
let a = wheel("alpha-pkg", "1.0.0");
land(&up, WHEEL_A, &a);
land(&down, WHEEL_A, &a);
let inv = up
.hub()
.invalidate(
"sec-anna",
"local",
WHEEL_A,
None,
Reason::CveAfter,
"CVE-2026-4242",
)
.unwrap();
// The manifest door needs an identity; the board door an Auditor.
assert_eq!(
request(
&up.http,
"GET",
"/-/sync/manifest?downstream=onprem&since=0",
None,
&[]
)
.status,
401
);
assert_eq!(
request(&up.http, "GET", "/-/sync/board", Some(&token("ida")), &[]).status,
403
);
let hd = down.hub();
let peer = hd.config().upstreams[0].clone();
let link = Recording::new();
let applied = hd.pull_once(&link, &peer).unwrap();
assert!(applied.ack_sent, "the ack was posted back");
// HOL-PRODUCTS-5: what a real downstream sends. The ack goes with the downstream's own
// credential (its `token_file`), as the manifest pull does.
let seen = link.seen.lock().unwrap().clone();
let ack = seen
.iter()
.find(|(m, u, _)| m == "POST" && u.ends_with("/-/sync/ack"))
.expect("the pull posted its ack");
assert_eq!(
ack.2.as_deref(),
Some(token("onprem").as_str()),
"the ack carries the downstream's bearer: {seen:?}"
);
// The ack door wants both: an authenticated identity AND the ack's own signature by an
// enrolled downstream. The real ack without a credential is a 401; a credential with a
// document that is no signed ack is refused all the same.
let real = applied.ack.to_bytes();
let anon = request(&up.http, "POST", "/-/sync/ack", None, &real);
assert_eq!(anon.status, 401, "no credential, no ack: {}", anon.text());
let forged = request(
&up.http,
"POST",
"/-/sync/ack",
Some(&token("onprem")),
br#"{"not":"a signed ack"}"#,
);
assert!(
(400..500).contains(&forged.status) && forged.text().contains("[holger-sync-"),
"a credential does not make an unsigned ack good: {} {}",
forged.status,
forged.text()
);
let row = include_str!("../src/exposed/http.rs")
.lines()
.find(|l| l.starts_with("/// | `POST /-/sync/ack` |"))
.expect("the sync doors' table has a row for POST /-/sync/ack")
.to_string();
assert!(
row.contains(
"| any authenticated identity, and the ack's own signature by an enrolled downstream |"
),
"the table must say both checks the door makes: {row}"
);
assert_eq!(applied.quarantined, vec![WHEEL_A.to_string()]);
assert_eq!(simple_page(&down, "alpha-pkg").status, 404);
let board = request(&up.http, "GET", "/-/sync/board", Some(&token("aud")), &[]);
assert_eq!(board.status, 200, "{}", board.text());
let b = board.json();
assert_eq!(b["schema"], "holger.sync-board/v1");
assert_eq!(b["alarm"], false);
let d = &b["downstreams"][0];
assert_eq!(d["name"], "onprem");
assert_eq!(d["state"], "current", "{b}");
assert_eq!(d["last_generation_applied"], b["generation"]);
assert_eq!(d["invalidations_reached"], 1);
assert_eq!(d["carrier"], "pull");
assert_eq!(b["invalidations"][0]["id"], inv.invalidation.id.as_str());
// Reached: no alarm, however late it is asked.
assert!(up.hub().check_alarms(now() + 100_000).unwrap().is_empty());
// A second pull with nothing new is a replay: acknowledged again, nothing held again.
let again = hd.pull_once(&HttpLink::new(), &peer).unwrap();
assert!(again.quarantined.is_empty());
assert_eq!(again.already_held, vec![WHEEL_A.to_string()]);
}
// ── the alarm ───────────────────────────────────────────────────────────────
#[test]
fn an_alarm_fires_when_no_ack_arrives_in_time() {
let k = keys();
let up = start(&upstream_ron(&k, 5));
let a = wheel("alpha-pkg", "1.0.0");
land(&up, WHEEL_A, &a);
let hu = up.hub();
let inv = hu
.invalidate(
"sec-anna",
"local",
WHEEL_A,
None,
Reason::Pulled,
"yanked by its publisher",
)
.unwrap();
// Nothing ever comes back from `onprem`.
let alarms = hu.check_alarms(inv.invalidation.issued_at + 6).unwrap();
assert_eq!(alarms.len(), 1);
assert!(
alarms[0]
.line()
.starts_with("ALARM sync: downstream 'onprem'"),
"{}",
alarms[0].line()
);
let board = request(&up.http, "GET", "/-/sync/board", Some(&token("aud")), &[]);
let b = board.json();
assert_eq!(b["alarm"], true, "{b}");
assert_eq!(b["downstreams"][0]["state"], "alarm");
assert_eq!(b["downstreams"][0]["pending"][0]["overdue"], true);
// The HTTP door issues an invalidation for a Security Approver only.
let refused = request(
&up.http,
"POST",
&format!("/-/sync/invalidate?coordinate={WHEEL_A}&reason=policy&detail=x"),
Some(&token("aud")),
&[],
);
assert_eq!(refused.status, 403, "{}", refused.text());
}
// ── the server's own sync job ───────────────────────────────────────────────
/// With the job queue, a pull-mode downstream pulls by itself every
/// `sync.tick_secs` (a `sync.tick` jera job): an invalidation on the upstream
/// reaches it, and its ack reaches the upstream, with nobody calling anything.
#[cfg(feature = "jobs")]
#[test]
fn the_tick_job_pulls_and_acknowledges_by_itself() {
let k = keys();
let up = start(&upstream_ron(&k, 3600));
let tok = tempfile::NamedTempFile::new().unwrap();
std::fs::write(tok.path(), token("onprem")).unwrap();
let down = start(&format!(
"{}, tick_secs: 1",
downstream_ron(
&k,
&format!(
r#", url: Some("http://{}"), token_file: Some("{}")"#,
up.http,
tok.path().display()
)
)
));
let a = wheel("alpha-pkg", "1.0.0");
land(&up, WHEEL_A, &a);
land(&down, WHEEL_A, &a);
up.hub()
.invalidate(
"sec-anna",
"local",
WHEEL_A,
None,
Reason::Malware,
"found later",
)
.unwrap();
let t = std::time::Instant::now();
while down.chain().state(&id(WHEEL_A)) != ArtifactState::Quarantined {
assert!(
t.elapsed() < Duration::from_secs(30),
"the tick never pulled the invalidation"
);
std::thread::sleep(Duration::from_millis(100));
}
while up.hub().board(now()).unwrap().downstreams[0].invalidations_reached != 1 {
assert!(
t.elapsed() < Duration::from_secs(30),
"the tick's ack never reached the upstream"
);
std::thread::sleep(Duration::from_millis(100));
}
}