server/lib/src/replication.rs
//! Repository REPLICATION — copy a repo's artifacts from a SOURCE holger store
//! into a DESTINATION store, content-id idempotent (parity §I,
//! `.nornir/replication-design.md`).
//!
//! The engine is a **pure function** over two backend sets: hand it a list of
//! `(source_backend, dest_backend)` repo PAIRS and a [`ReplicationOptions`], and
//! it walks each source's `list()`, `fetch()`es the bytes, and — for every
//! artifact the destination does not already hold *by content id* — `put()`s it
//! into the writable destination. Its only I/O is the backend calls it is given,
//! so it is unit-testable with in-memory backends and reused verbatim by both
//! surfaces: the `holger-server replicate --from --to` CLI verb (two separate
//! file-backed stores) and the `ReplicationService.Replicate` gRPC RPC (two repos
//! within one running server).
//!
//! **Read-from-source / write-to-dest, never the reverse.** The source is only
//! ever `list()`ed + `fetch()`ed; the destination is only ever `fetch()`ed (to
//! test presence) + `put()` (to copy). A source is never mutated.
//!
//! **Idempotent by content id.** The dedup key is the artifact's content digest
//! (lowercase-hex sha256 of the bytes, the same digest the search engine matches).
//! Before copying, the engine fetches the destination's bytes for the same
//! coordinate: if they are already present AND hash to the same content id, the
//! artifact is *skipped*. So a second replicate run over an unchanged source is a
//! pure no-op (every artifact skipped, nothing written) — the property the
//! executor test asserts. A coordinate whose destination content DIFFERS is not
//! "present by content id", so it is (re-)copied, converging the destination onto
//! the source.
//!
//! **Gated.** `execute` defaults to `false` — a dry run reports what *would* copy
//! and writes nothing; an actual copy only happens under
//! [`ReplicationOptions::executing`]. A destination backend that is not writable
//! fails that pair's artifacts *closed* (reported, never silently dropped).
use std::path::PathBuf;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use traits::{ArtifactId, RepositoryBackendTrait};
// The house engine, not rayon (ROOT LAW #0). `znippy_zoomies` re-exports gatling's
// modules unchanged, so this is the same `rotaryengine` every other nordisk repo
// fans out on -- and it is now CALLED by holger-server, which stinky-01 §1.1
// measured as linking it and calling it zero times.
use znippy_zoomies::gatling_forkjoin::gatling_map_balanced;
/// Lowercase-hex sha256 — the content id replication dedups on. Kept local (the
/// same choice `search.rs` made) so the engine is self-contained + testable in
/// isolation.
fn content_id(data: &[u8]) -> String {
// LAW #5 dedup: consume the shared `nornir-hash` leaf (edda) so the replication
// content id is byte-identical to search.rs / the backends' recompute.
nornir_hash::sha256_hex(data)
}
/// Knobs for a replication run.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
pub struct ReplicationOptions {
/// When `false` (the DEFAULT), a dry run: report what *would* copy, write
/// nothing. An actual `put()` into the destination only happens when `true`.
pub execute: bool,
/// Per-pair cap on artifacts examined; `0` ⇒ unlimited. Bounds a huge source.
pub limit: usize,
}
impl ReplicationOptions {
/// A dry-run preview (writes nothing), unbounded.
pub fn preview() -> Self {
Self::default()
}
/// An executing run (actually copies), unbounded.
pub fn executing() -> Self {
Self {
execute: true,
limit: 0,
}
}
}
/// What happened to a single source artifact under replication.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum Outcome {
/// Copied into the destination (an executing run wrote the bytes).
Copied,
/// Already present in the destination by content id — skipped (idempotent).
Skipped,
/// A dry run that would copy this artifact (nothing was written).
WouldCopy,
/// Could not replicate (source unreadable, destination read-only, put failed).
Failed,
}
/// One source artifact's replication result, tagged with both repo names + its
/// content id.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReplicatedArtifact {
pub source_repo: String,
pub dest_repo: String,
pub id: ArtifactId,
/// Lowercase-hex sha256 of the source bytes (empty when the source could not
/// be read).
pub content_id: String,
pub size_bytes: i64,
pub outcome: Outcome,
/// Human detail — the failure reason for [`Outcome::Failed`], else empty.
pub detail: String,
}
/// The result of a [`replicate_pairs`] run: one row per source artifact examined,
/// plus derived tallies.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReplicationReport {
pub items: Vec<ReplicatedArtifact>,
/// True when a pair hit its `limit` cap and more source artifacts remained.
pub truncated: bool,
}
impl ReplicationReport {
fn count(&self, o: Outcome) -> usize {
self.items.iter().filter(|i| i.outcome == o).count()
}
/// Artifacts actually written into the destination.
pub fn copied(&self) -> usize {
self.count(Outcome::Copied)
}
/// Artifacts already present by content id (skipped).
pub fn skipped(&self) -> usize {
self.count(Outcome::Skipped)
}
/// Artifacts a dry run would copy.
pub fn would_copy(&self) -> usize {
self.count(Outcome::WouldCopy)
}
/// Artifacts that could not be replicated.
pub fn failed(&self) -> usize {
self.count(Outcome::Failed)
}
/// Total source artifacts examined.
pub fn examined(&self) -> usize {
self.items.len()
}
/// True when nothing failed — the replication is fully accounted for.
pub fn is_complete(&self) -> bool {
self.failed() == 0
}
/// Sum of bytes actually copied (0 on a dry run).
pub fn copied_bytes(&self) -> i64 {
self.items
.iter()
.filter(|i| i.outcome == Outcome::Copied)
.map(|i| i.size_bytes)
.sum()
}
}
/// One source→destination repository pairing for a replication run. The source is
/// read-only here; the destination is where artifacts land.
pub struct RepoPair {
pub source_name: String,
pub source: Arc<dyn RepositoryBackendTrait>,
pub dest_name: String,
pub dest: Arc<dyn RepositoryBackendTrait>,
}
/// Replicate every pairing in `pairs`, source→dest, content-id idempotent. Pure
/// over the backends it is given (the only I/O is their `list`/`fetch`/`put`), so
/// both the CLI and the gRPC RPC drive it unchanged. A source that cannot be
/// listed contributes no rows for that pair (best-effort, logged); every artifact
/// that *is* listed yields exactly one [`ReplicatedArtifact`] row.
pub fn replicate_pairs(pairs: &[RepoPair], opts: &ReplicationOptions) -> ReplicationReport {
let mut report = ReplicationReport::default();
for pair in pairs {
replicate_one_pair(pair, opts, &mut report);
}
report
}
/// Peak bytes phase A is allowed to hold in RAM at once.
///
/// The fan-out reads artifacts ahead of the committer, so without a cap a
/// 100 000-crate source would have every crate resident before the first `put`.
/// Entries are therefore batched into WINDOWS whose declared sizes sum to at
/// most this, and each window is read in parallel and committed before the next
/// is read. With 1 KiB crates that is a window of a quarter-million units (the
/// fan-out saturates); with 1 GiB artifacts it is one unit (the fan-out
/// degenerates to the old serial loop, which is the correct answer for memory).
/// stinky-01 §7.4 is the reason the cap exists at all: nothing else in this
/// process accounts for resident artifact bytes.
const REPLICATION_WINDOW_BYTES: u64 = 256 * 1024 * 1024;
/// What phase A decided about one artifact. Phase B turns it into a report row.
enum Prepared {
/// The source listed it but could not produce bytes. One `Failed` row.
SourceFailed(String),
/// The destination already holds these exact bytes. One `Skipped` row.
Present { cid: String, size: i64 },
/// Needs copying, and the bytes are in hand.
Needs {
bytes: traits::Bytes,
cid: String,
size: i64,
},
}
/// Replicate a single source→dest pair, appending its rows to `report`.
///
/// **Two phases — the fan-out shape this house calls gatling** (skade
/// `parquet_io.rs:131`, znippy-zoomies `ljar/src/batch.rs:113`): the units are
/// known up front (`list()` hands them over, with `size_bytes` as the weight),
/// the read half is pure, and the write half has to stay serial because a
/// writable znippy destination takes ONE process-wide lock per `put`.
///
/// * **Phase A — read, parallel.** `gatling_map_balanced` over the window,
/// weighted by `size_bytes` so the 40 MB artifact is scheduled before the
/// 4 KB one (heaviest-first, no barrier). Each unit does `source.fetch` →
/// `content_id` → the destination presence probe. All three are read-only on
/// both backends, so no unit can race another. Results come back in the
/// original index order.
/// * **Phase B — write, serial.** One committer walks phase A's results in
/// order and `put`s the ones that need copying, so the report rows are in
/// exactly the order the old serial loop produced them.
///
/// What this replaces: the whole thing was ONE `for` loop — fetch, sha256 and
/// presence probe on one core with the other 31 idle, and sha256 runs at
/// **0.61 GB/s on one core** (stinky-01 §2.2 MEASURED), so a mirror of N
/// artifacts cost N × (fetch + hash + probe + put) of wall time in series. A
/// dry run (`execute = false`) never reaches the serial half at all.
fn replicate_one_pair(pair: &RepoPair, opts: &ReplicationOptions, report: &mut ReplicationReport) {
let mut entries = match pair.source.list(None, 0) {
Ok(e) => e,
Err(e) => {
log::warn!(
"replicate: list failed in source repo '{}': {e}",
pair.source_name
);
return;
}
};
let cap = if opts.limit == 0 {
usize::MAX
} else {
opts.limit
};
if entries.len() > cap {
entries.truncate(cap);
report.truncated = true;
}
let dest_writable = pair.dest.is_writable();
let mut offset = 0usize;
while offset < entries.len() {
// Cut the next window at the byte budget, always at least one unit so a
// single artifact larger than the whole budget still makes progress.
let mut window_bytes = 0u64;
let mut end = offset;
while end < entries.len() {
let sz = entries[end].size_bytes.max(0) as u64;
if end > offset && window_bytes.saturating_add(sz) > REPLICATION_WINDOW_BYTES {
break;
}
window_bytes = window_bytes.saturating_add(sz);
end += 1;
}
let window = &entries[offset..end];
// ── Phase A: fetch + content id + presence probe, one unit per artifact,
// heaviest first, index-ordered out. ──
let prepared = gatling_map_balanced(
window,
0,
1,
|e: &traits::ArtifactEntry| e.size_bytes.max(0) as u64,
|_i, entry: &traits::ArtifactEntry| {
let id = &entry.id;
// Read the SOURCE bytes (the source is only ever read).
let bytes = match pair.source.fetch(id) {
Ok(Some(b)) => b,
// Listed but unreadable — record a failure, never a silent drop.
Ok(None) => {
return Prepared::SourceFailed(
"source listed the artifact but fetch returned no bytes".to_string(),
);
}
Err(e) => return Prepared::SourceFailed(format!("source fetch failed: {e}")),
};
let cid = content_id(&bytes);
let size = bytes.len() as i64;
// Idempotency: already present in the DESTINATION by content id?
let already_present = match pair.dest.fetch(id) {
Ok(Some(existing)) => content_id(&existing) == cid,
// Absent, or destination read error → treat as "needs copy" (a
// read error on the presence probe must not skip the copy).
Ok(None) => false,
Err(e) => {
log::warn!(
"replicate: dest presence probe failed in '{}' for {}@{}: {e}",
pair.dest_name,
id.name,
id.version
);
false
}
};
if already_present {
Prepared::Present { cid, size }
} else {
Prepared::Needs { bytes, cid, size }
}
},
);
// ── Phase B: the single committer. ──
for (entry, decision) in window.iter().zip(prepared) {
let id = &entry.id;
match decision {
Prepared::SourceFailed(why) => {
report
.items
.push(row(pair, id, "", 0, Outcome::Failed, &why));
}
Prepared::Present { cid, size } => {
report
.items
.push(row(pair, id, &cid, size, Outcome::Skipped, ""));
}
// Needs copy. A read-only destination fails CLOSED (reported,
// never silently dropped).
Prepared::Needs { cid, size, .. } if !dest_writable => {
report.items.push(row(
pair,
id,
&cid,
size,
Outcome::Failed,
&format!("destination repo '{}' is read-only", pair.dest_name),
));
}
// Dry run: report the intent, write nothing.
Prepared::Needs { cid, size, .. } if !opts.execute => {
report
.items
.push(row(pair, id, &cid, size, Outcome::WouldCopy, ""));
}
// Executing: copy the bytes into the destination.
Prepared::Needs { bytes, cid, size } => match pair.dest.put(id, &bytes) {
Ok(()) => report
.items
.push(row(pair, id, &cid, size, Outcome::Copied, "")),
Err(e) => report.items.push(row(
pair,
id,
&cid,
size,
Outcome::Failed,
&format!("destination put failed: {e}"),
)),
},
}
}
offset = end;
}
}
/// Build one report row (keeps `replicate_one_pair` readable).
fn row(
pair: &RepoPair,
id: &ArtifactId,
cid: &str,
size: i64,
outcome: Outcome,
detail: &str,
) -> ReplicatedArtifact {
ReplicatedArtifact {
source_repo: pair.source_name.clone(),
dest_repo: pair.dest_name.clone(),
id: id.clone(),
content_id: cid.to_string(),
size_bytes: size,
outcome,
detail: detail.to_string(),
}
}
// ═══════════════════════════════════════════════════════════════════════════
// THE ESTATE — what `replicate_pairs` alone can NEVER make a twin out of
// ═══════════════════════════════════════════════════════════════════════════
//
// Plan §3 W10, `.nornir/stinky/02-twin-growth.md` S2-4, MEASURED on `main` at
// `2fb3602`: `replicate_pairs` moves artifacts **by coordinate**. That is the
// right engine and it is not enough. A box built from it alone holds no
// evidence store, no audit segments, no properties, no SBOMs, no promoted
// (sealed) store, no update-bundle state and no served config — so promoted, it
// serves artifacts with NO provenance behind them and a history that begins at
// the promotion. gunnar's un-metered-twin scar in a different coat: the failover
// keeps serving and silently stops being able to prove anything.
//
// `replicate_estate` is the second half, and it is deliberately BESIDE
// `replicate_pairs` rather than inside it: `holger-traits` is PUBLISHED and the
// existing surface (`RepoPair`, `ReplicationOptions`, `ReplicationReport`,
// `replicate_pairs`) is untouched, so every current caller — the CLI verb and
// the `ReplicationService.Replicate` RPC — compiles and behaves exactly as
// before.
//
// **Same two properties as the artifact half.** Idempotent by CONTENT ID (the
// same lowercase-hex sha256), so a second run over an unchanged primary copies
// nothing; and gated by `ReplicationOptions::execute`, so a dry run reports
// what it would move and writes not one byte.
//
// **Read the primary, write the twin, never the reverse.** Nothing in here
// opens the primary's tree for writing.
/// One named item of the data root a twin must hold, and WHY — the reason
/// travels with the name so a report and a refusal can both say what is
/// missing rather than printing a directory.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EstateItem {
/// The path under the data root, exactly as the store that owns it builds it.
pub path: &'static str,
/// What a twin loses without it.
pub what: &'static str,
/// `true` for a single file (the served config), `false` for a store tree.
pub is_file: bool,
}
/// **Everything under `<data_root>` that makes a promoted twin able to prove
/// what it serves.** Each entry names the store that builds the path, so a
/// renamed store is a one-line change here and a failing test, not a silently
/// un-replicated store.
///
/// Absent on the primary is NOT an error: a deployment that hosts no SBOMs has
/// no `holger-sboms`, and a twin of it correctly holds none. The report says
/// `absent` by name so "nothing to copy" and "copied nothing" are never the
/// same sentence.
pub const ESTATE: &[EstateItem] = &[
EstateItem {
path: "holger-attestations",
what: "the evidence/attestation store (DSSE envelopes) — without it a promoted twin \
serves artifacts nothing has ever attested",
is_file: false,
},
EstateItem {
path: "holger-audit",
what: "the audit segments — append-only Arrow IPC, hash-chained; without them the \
twin's history begins at its promotion",
is_file: false,
},
EstateItem {
path: "holger-props",
what: "custom properties — the retention pin and the promotion engine's required \
property both live here",
is_file: false,
},
EstateItem {
path: "holger-sboms",
what: "hosted SBOMs",
is_file: false,
},
EstateItem {
path: "holger-prod-store",
what: "the sealed PROD store and its promotion lineage — the content-addressed refs \
that make a coordinate promoted at all",
is_file: false,
},
EstateItem {
path: "holger-update-state",
what: "the update-bundle store: the anti-rollback sequence state a twin must not \
come up behind",
is_file: false,
},
EstateItem {
path: "holger-import-receipts",
what: "import receipts — what this installation has accepted",
is_file: false,
},
EstateItem {
path: "plugins",
what: "the plugin plane's private store, including the anti-rollback floor per \
(supplier, product)",
is_file: false,
},
// ── added 2026-09-21 after an ENUMERATION of every path this server builds
// from its data root (lane T5). Each of these was a store the primary
// writes and the twin never saw, which is the whole failure mode the
// owner's "the twin has config + data" order names.
EstateItem {
path: "holger-passkeys",
what: "the console's identities — `credentials.json` (every registered passkey), \
`invites.json` and the session MAC key. Without them a promoted twin has NO \
ONE who can log into it and every outstanding invite is void",
is_file: false,
},
EstateItem {
path: "holger-entitlements",
what: "the metering plane's entitlement store (`metered::STORE_DIR`) — the \
un-metered-twin scar in holger's shape: a promoted twin that does not hold \
this enforces nothing and says nothing",
is_file: false,
},
EstateItem {
path: "meters",
what: "the plugin plane's meter readings (`DirStore` under `<data_root>/meters`) — \
what has been consumed, which a promotion must not reset to zero",
is_file: false,
},
EstateItem {
path: "holger-jobs",
what: "the jera job ledger (`holger-jobs/jobs.redb`) — the recent-operations list \
every promotion is routed through",
is_file: false,
},
EstateItem {
path: "jobs",
what: "the durable job queue (`jobs::JobQueue`, `<data_root>/jobs`) — queued and \
in-flight work, including its `staging/`",
is_file: false,
},
// ── added 2026-09-29 (the non-payment ruling): a repository deleted on the
// main must be deleted on the twin too, and the twin's mirror never
// deletes — so the main's purge records travel with the estate and the
// twin applies them (`purge::apply_records`). The twin's memory of what it
// applied (`purge::APPLIED_FILE`) is deliberately NOT an item.
EstateItem {
path: "holger-purges",
what: "the purge records (`purge::RECORDS_DIR`) — one immutable file per repository \
the main emptied. Without them a purged tenant's artifacts survive on the \
twin, whose mirror never deletes, and come back with a recharge",
is_file: false,
},
// ── added 2026-10-21 (lane P4, A8 lane 10): what a SALE leaves on the main.
// Gunnar's sold state rides his twin and comes back with a rebuild; without
// these a twin follows none of the repositories sold after its config was
// written, and a recharged main forgets which signup owns which repository.
// The rest of `fleet/` (the pull channel's own room want) is this box's own.
EstateItem {
path: "fleet/signups.json",
what: "the sign-ups bound to their repositories (`fleet_box::SIGNUPS_REL`) — \
without it a recharged main lets another sign-up take one that is bound",
is_file: true,
},
EstateItem {
path: "fleet/sold-repositories.json",
what: "the repositories sign-ups made, with what each is shaped like — \
without it the twin follows none of them and a recharged main serves none of them",
is_file: true,
},
// ── added 2026-10-28 (lane OX-WGDS2, the owner's lifecycle rule): a dead main is
// rebuilt from its twin, and its downstreams must not notice. The twin's follow
// never deletes, so a revoked downstream's old record may stay beside its
// revocation; the revocation is what `sync::SyncHub::downstreams` reads.
EstateItem {
path: "holger-sync",
what: "the sync state (`sync::STATE_DIR`): the downstreams the console enrolled, their \
revocations, the WireGuard addresses, the ledger's generations and the sync key \
every downstream trusts. Without it a rebuilt main has no downstream, a revoked \
one could enrol again unseen, and every manifest it signs is refused",
is_file: false,
},
EstateItem {
path: "holger.ron",
what: "the served config — `[estate-twin-drift]` demands the twin's config be \
BYTE-IDENTICAL to the primary's; the role is applied at wiring time, never \
by editing the file",
is_file: true,
},
];
/// What happened to one estate item.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct EstateItemReport {
pub path: String,
pub what: String,
/// The primary does not have this item at all — nothing to copy, and that
/// is a normal state, not a failure.
pub absent_on_primary: bool,
pub copied: usize,
pub skipped: usize,
pub would_copy: usize,
pub failed: usize,
pub copied_bytes: i64,
/// Failure detail, per failed file, newest last. Empty when nothing failed.
pub failures: Vec<String>,
}
impl EstateItemReport {
fn new(item: &EstateItem) -> Self {
Self {
path: item.path.to_string(),
what: item.what.to_string(),
absent_on_primary: false,
copied: 0,
skipped: 0,
would_copy: 0,
failed: 0,
copied_bytes: 0,
failures: Vec::new(),
}
}
}
/// The result of a [`replicate_estate`] run — one row per [`ESTATE`] item.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct EstateReport {
pub items: Vec<EstateItemReport>,
/// **Everything the primary's data root holds that [`ESTATE`] does not
/// name** — by name, every run, sorted.
///
/// This field exists because the failure it catches is silent by
/// construction. `ESTATE` is a hand-written list; a store added to the
/// server next month lands beside it on the primary's disk and is simply
/// never copied, and every report before this field said `complete`. The
/// biggest occupant of a live holger data root is the ARTIFACT STORE
/// itself (`/var/lib/holger/store` on holger.rs) — config-named, so no
/// fixed entry can cover it, and it is moved by
/// [`replicate_pairs`]/`holger-server replicate` rather than by this
/// function. A twin that copied the evidence and not the artifacts must say
/// so out loud rather than report a clean run.
#[serde(default)]
pub unreplicated: Vec<String>,
}
impl EstateReport {
pub fn copied(&self) -> usize {
self.items.iter().map(|i| i.copied).sum()
}
pub fn skipped(&self) -> usize {
self.items.iter().map(|i| i.skipped).sum()
}
pub fn would_copy(&self) -> usize {
self.items.iter().map(|i| i.would_copy).sum()
}
pub fn failed(&self) -> usize {
self.items.iter().map(|i| i.failed).sum()
}
/// True when nothing failed — every file is accounted for.
pub fn is_complete(&self) -> bool {
self.failed() == 0
}
/// True when the run wrote nothing and would have written nothing: the
/// no-op a second replicate over an unchanged primary must be.
pub fn is_noop(&self) -> bool {
self.copied() == 0 && self.would_copy() == 0 && self.failed() == 0
}
/// The row for one item, by path.
pub fn item(&self, path: &str) -> Option<&EstateItemReport> {
self.items.iter().find(|i| i.path == path)
}
/// The named warning for [`EstateReport::unreplicated`], or `None` when the
/// twin holds everything the primary's data root contains. `holger-twin-3`,
/// written in ONE place so the log, a door and a test say the same thing.
///
/// A WARNING and never a refusal, for the same reason `holger-twin-2` is:
/// an incomplete twin is still a restore source, and refusing here would
/// take the only copy offline.
pub fn unreplicated_warning(&self) -> Option<String> {
if self.unreplicated.is_empty() {
return None;
}
Some(holger_errcode::TWIN_ESTATE_INCOMPLETE.tag(format!(
"the primary's data root holds {} item(s) this replication does not cover: {}. \
A promotion of this twin would come up without them. The artifact store is \
moved by `holger-server replicate --from --to`, not by the estate pass; \
anything else on this list is a store nobody has taught the twin about.",
self.unreplicated.len(),
self.unreplicated.join(", ")
)))
}
}
/// **Every child of `primary_root` that [`ESTATE`] does not name.** Sorted, so
/// the report is stable and a test can assert on it.
///
/// Dotfiles are excluded: `.holger-import.lock` and `.holger-import.staged.znippy`
/// are a running import's scratch, not state a twin should hold, and copying a
/// lock file to a follower would be actively wrong.
pub fn unreplicated_children(primary_root: &std::path::Path) -> Vec<String> {
let named: std::collections::HashSet<&str> = ESTATE.iter().map(|i| i.path).collect();
let mut out: Vec<String> = match std::fs::read_dir(primary_root) {
Ok(rd) => rd
.filter_map(|e| e.ok())
.filter_map(|e| e.file_name().into_string().ok())
.filter(|n| !n.starts_with('.'))
.filter(|n| !named.contains(n.as_str()))
// ★ THE ROLE RECORD IS NEVER REPLICATED, and it is not "uncovered"
// either — it is excluded ON PURPOSE. Copying it would turn a
// restored primary into a twin of the box that just died: the
// restore's destination would come up refusing every write by name,
// pointing at a machine that no longer exists. A box's role is that
// box's own bootstrap fact and travels with nothing.
.filter(|n| n != crate::twin::ROLE_RECORD)
// ★ …and neither is the box's replication IDENTITY: its issuer (a
// twin holding its primary's issuer key could mint the primary's
// twins) and its reader credentials. Excluded, not uncovered — and
// so not what makes a box un-fresh either: a box opens its
// replication door, minting the issuer, on its FIRST start, before
// any election could have found it fresh (`crate::twin_tls::BOX_OWN`).
.filter(|n| !crate::twin_tls::BOX_OWN.contains(&n.as_str()))
.collect(),
// An unreadable primary root is not this function's failure to report:
// `replicate_estate` already turns that into rows.
Err(_) => Vec::new(),
};
out.sort();
out
}
/// **Replicate the whole estate of a data root**, primary → twin: every item of
/// [`ESTATE`], content-id idempotent, gated by [`ReplicationOptions::execute`].
///
/// This is the side-store half of making a twin. The artifact half is
/// [`replicate_pairs`], which is unchanged and still the only thing that moves
/// bytes by coordinate; a follower runs both (`crate::twin::follow_once`).
///
/// `Err` is reserved for a failure of the RUN — a twin root that cannot be
/// created. A single unreadable file inside a store is a `failed` ROW naming
/// it, never a silent drop and never an aborted run: a twin that copied nine of
/// ten audit segments must say which one it missed.
pub fn replicate_estate(
primary_root: &std::path::Path,
twin_root: &std::path::Path,
opts: &ReplicationOptions,
) -> Result<EstateReport, String> {
if opts.execute {
std::fs::create_dir_all(twin_root)
.map_err(|e| format!("creating the twin data root {}: {e}", twin_root.display()))?;
}
let mut report = EstateReport::default();
for item in ESTATE {
report
.items
.push(replicate_item(primary_root, twin_root, item, opts));
}
report.unreplicated = unreplicated_children(primary_root);
if let Some(warning) = report.unreplicated_warning() {
log::warn!("{warning}");
}
Ok(report)
}
/// Every regular file under `dir`, as (absolute path, path relative to `dir`,
/// declared size). Symlinks are NOT followed: a twin must hold the estate, not
/// a link into the primary's filesystem, and following one would silently make
/// the twin depend on a volume it does not own.
pub(crate) fn walk_files(dir: &std::path::Path) -> std::io::Result<Vec<(PathBuf, PathBuf, u64)>> {
let mut out = Vec::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(d) = stack.pop() {
for entry in std::fs::read_dir(&d)? {
let entry = entry?;
let path = entry.path();
let meta = entry.metadata()?;
if meta.is_symlink() {
continue;
}
if meta.is_dir() {
stack.push(path);
} else if meta.is_file() {
let rel = path
.strip_prefix(dir)
.expect("walked below dir")
.to_path_buf();
out.push((path, rel, meta.len()));
}
}
}
// Stable order, so two runs produce comparable reports.
out.sort_by(|a, b| a.1.cmp(&b.1));
Ok(out)
}
/// What phase A decided about one estate file.
enum EstatePrepared {
/// The source file could not be read. One `failed` row.
SourceFailed(String),
/// The twin already holds these exact bytes.
Present,
/// Needs copying, bytes in hand.
Needs(Vec<u8>),
}
fn replicate_item(
primary_root: &std::path::Path,
twin_root: &std::path::Path,
item: &EstateItem,
opts: &ReplicationOptions,
) -> EstateItemReport {
let mut row = EstateItemReport::new(item);
let src_base = primary_root.join(item.path);
let dest_base = twin_root.join(item.path);
let files: Vec<(PathBuf, PathBuf, u64)> = if item.is_file {
match std::fs::metadata(&src_base) {
Ok(m) if m.is_file() => vec![(src_base.clone(), PathBuf::new(), m.len())],
_ => {
row.absent_on_primary = true;
return row;
}
}
} else {
match walk_files(&src_base) {
// The audit INDEX (`holger-audit/index/`, lane HOL-AUDIT) is derived from the
// chain and rewritten in place: every box builds its own from the replicated
// `audit.jsonl`, so it is never copied (a copy would race its temp files).
Ok(f) if item.path == "holger-audit" => f
.into_iter()
.filter(|(_, rel, _)| !rel.starts_with(crate::audit::arrow_ipc::INDEX_DIR))
.collect(),
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
row.absent_on_primary = true;
return row;
}
Err(e) => {
row.failed += 1;
row.failures
.push(format!("reading {}: {e}", src_base.display()));
return row;
}
}
};
let mut offset = 0usize;
while offset < files.len() {
// The same window discipline the artifact half uses: cut at the byte
// budget, always at least one unit, so a store of large segments never
// becomes resident all at once. stinky-01 §7.4 — nothing else in this
// process accounts for resident bytes.
let mut window_bytes = 0u64;
let mut end = offset;
while end < files.len() {
let sz = files[end].2;
if end > offset && window_bytes.saturating_add(sz) > REPLICATION_WINDOW_BYTES {
break;
}
window_bytes = window_bytes.saturating_add(sz);
end += 1;
}
let window = &files[offset..end];
// ── Phase A: read + content id + presence probe, in parallel, heaviest
// first. The house engine, not rayon (ROOT LAW #0). Both ends are only
// READ here, so no unit can race another. ──
let prepared = gatling_map_balanced(
window,
0,
1,
|f: &(PathBuf, PathBuf, u64)| f.2,
|_i, (src, rel, _sz): &(PathBuf, PathBuf, u64)| {
let bytes = match std::fs::read(src) {
Ok(b) => b,
Err(e) => {
return EstatePrepared::SourceFailed(format!(
"reading {}: {e}",
src.display()
))
}
};
let dest = if rel.as_os_str().is_empty() {
dest_base.clone()
} else {
dest_base.join(rel)
};
// Idempotency by content id, exactly as the artifact half: the
// twin already holding these BYTES is a skip; holding different
// bytes at the same path is not "present", so it is re-copied
// and the twin converges onto the primary.
let present = match std::fs::read(&dest) {
Ok(existing) => content_id(&existing) == content_id(&bytes),
Err(_) => false,
};
if present {
EstatePrepared::Present
} else {
EstatePrepared::Needs(bytes)
}
},
);
// ── Phase B: the single committer. ──
for ((src, rel, _sz), decision) in window.iter().zip(prepared) {
let dest = if rel.as_os_str().is_empty() {
dest_base.clone()
} else {
dest_base.join(rel)
};
match decision {
EstatePrepared::SourceFailed(why) => {
row.failed += 1;
row.failures.push(why);
}
EstatePrepared::Present => row.skipped += 1,
EstatePrepared::Needs(bytes) if !opts.execute => {
row.would_copy += 1;
let _ = (src, bytes);
}
EstatePrepared::Needs(bytes) => {
let wrote = dest
.parent()
.map(std::fs::create_dir_all)
.unwrap_or(Ok(()))
// HOL-NCSC: the estate's state holds keys (the sync key, the passkey
// store, the WireGuard identity): written 0600.
.and_then(|()| crate::keyfile::write_private(&dest, &bytes));
match wrote {
Ok(()) => {
row.copied += 1;
row.copied_bytes += bytes.len() as i64;
}
Err(e) => {
row.failed += 1;
row.failures
.push(format!("writing {}: {e}", dest.display()));
}
}
}
}
}
offset = end;
}
row
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use std::sync::Mutex;
use traits::{ArtifactEntry, ArtifactFormat};
/// A mutable in-memory backend: `put` really stores (so a copied artifact is
/// afterwards `fetch`-able, and the round-trip is byte-checkable). `writable`
/// gates writes. `list`/`fetch`/`put` all go through the shared map.
struct MemRepo {
name: String,
writable: bool,
store: Mutex<Vec<(ArtifactId, Vec<u8>)>>,
}
impl MemRepo {
fn new(name: &str, writable: bool) -> Self {
Self {
name: name.into(),
writable,
store: Mutex::new(Vec::new()),
}
}
fn seeded(name: &str, entries: Vec<(ArtifactId, Vec<u8>)>) -> Self {
Self {
name: name.into(),
writable: false,
store: Mutex::new(entries),
}
}
fn arc(self) -> Arc<dyn RepositoryBackendTrait> {
Arc::new(self)
}
}
#[async_trait]
impl RepositoryBackendTrait for MemRepo {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
self.writable
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(self
.store
.lock()
.unwrap()
.iter()
.find(|(e, _)| e == id)
.map(|(_, b)| traits::Bytes::from(b.clone())))
}
fn put(&self, id: &ArtifactId, data: &[u8]) -> anyhow::Result<()> {
if !self.writable {
anyhow::bail!("read-only");
}
let mut s = self.store.lock().unwrap();
if let Some(slot) = s.iter_mut().find(|(e, _)| e == id) {
slot.1 = data.to_vec(); // overwrite (converge on source)
} else {
s.push((id.clone(), data.to_vec()));
}
Ok(())
}
fn list(&self, _name: Option<&str>, _limit: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
Ok(self
.store
.lock()
.unwrap()
.iter()
.map(|(id, b)| ArtifactEntry {
id: id.clone(),
size_bytes: b.len() as i64,
content_type: "application/octet-stream".into(),
// A test double knows neither, and says so.
..ArtifactEntry::UNKNOWN_PROVENANCE
})
.collect())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
fn id(name: &str, ver: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.into(),
version: ver.into(),
}
}
/// A source seeded with three crates; a fresh writable destination.
fn pair(
dest_writable: bool,
) -> (
Arc<dyn RepositoryBackendTrait>,
Arc<dyn RepositoryBackendTrait>,
) {
let src = MemRepo::seeded(
"rust-dev",
vec![
(id("serde", "1.0.0"), b"serde-bytes".to_vec()),
(id("serde", "1.0.1"), b"serde-newer".to_vec()),
(id("tokio", "1.40.0"), b"tokio-bytes".to_vec()),
],
)
.arc();
let dst = MemRepo::new("rust-mirror", dest_writable).arc();
(src, dst)
}
fn one_pair(
src: Arc<dyn RepositoryBackendTrait>,
dst: Arc<dyn RepositoryBackendTrait>,
) -> Vec<RepoPair> {
vec![RepoPair {
source_name: "rust-dev".into(),
source: src,
dest_name: "rust-mirror".into(),
dest: dst,
}]
}
#[test]
fn dry_run_reports_every_artifact_and_writes_nothing() {
let (src, dst) = pair(true);
let pairs = one_pair(src, dst.clone());
let r = replicate_pairs(&pairs, &ReplicationOptions::preview());
assert_eq!(r.would_copy(), 3, "all 3 source artifacts would copy");
assert_eq!(r.copied(), 0, "a dry run copies nothing");
// RED-when-broken: a dry run that actually wrote would leave the dest
// non-empty.
assert!(
dst.list(None, 0).unwrap().is_empty(),
"dry run must NOT write to the destination"
);
}
#[test]
fn execute_copies_every_source_artifact_byte_identical() {
let (src, dst) = pair(true);
let pairs = one_pair(src.clone(), dst.clone());
let r = replicate_pairs(&pairs, &ReplicationOptions::executing());
assert_eq!(r.copied(), 3, "all 3 copied");
assert!(r.is_complete(), "no failures");
// RED-when-broken CORE assertion: the destination holds EVERY source
// artifact, byte-identical. A dropped/garbled copy fails right here.
for entry in src.list(None, 0).unwrap() {
let want = src.fetch(&entry.id).unwrap().expect("source has it");
let got = dst
.fetch(&entry.id)
.unwrap()
.unwrap_or_else(|| panic!("destination MISSING {:?} after replicate", entry.id));
assert_eq!(
got, want,
"destination bytes must equal source bytes for {:?}",
entry.id
);
}
}
#[test]
fn rerun_is_a_no_op_everything_already_present_by_content_id() {
let (src, dst) = pair(true);
// First run copies everything.
let first = replicate_pairs(
&one_pair(src.clone(), dst.clone()),
&ReplicationOptions::executing(),
);
assert_eq!(first.copied(), 3);
// Second run over the unchanged source: NOTHING copies, all skipped.
let second = replicate_pairs(&one_pair(src, dst), &ReplicationOptions::executing());
assert_eq!(second.copied(), 0, "re-run must copy nothing (idempotent)");
assert_eq!(
second.skipped(),
3,
"every artifact already present by content id"
);
assert!(second.is_complete());
}
#[test]
fn changed_content_is_recopied_not_skipped() {
let (src, dst) = pair(true);
// Pre-seed the destination coordinate with DIFFERENT bytes than the source.
dst.put(&id("serde", "1.0.0"), b"stale-different-bytes")
.unwrap();
let r = replicate_pairs(
&one_pair(src.clone(), dst.clone()),
&ReplicationOptions::executing(),
);
// serde 1.0.0 differs → recopied; the other two are new → copied.
assert_eq!(
r.copied(),
3,
"the divergent coordinate is re-copied, not skipped"
);
assert_eq!(r.skipped(), 0);
// Destination converged onto the source content.
assert_eq!(
&dst.fetch(&id("serde", "1.0.0")).unwrap().unwrap()[..],
&b"serde-bytes"[..],
"destination now matches source bytes"
);
}
#[test]
fn read_only_destination_fails_closed_never_silently_drops() {
let (src, dst) = pair(false); // destination NOT writable
let r = replicate_pairs(
&one_pair(src, dst.clone()),
&ReplicationOptions::executing(),
);
// Every artifact is accounted for as a FAILURE, not silently skipped.
assert_eq!(
r.failed(),
3,
"a read-only destination fails every copy closed"
);
assert_eq!(r.copied(), 0);
assert!(!r.is_complete(), "failures ⇒ not complete");
assert!(
dst.list(None, 0).unwrap().is_empty(),
"nothing written to a read-only dest"
);
}
#[test]
fn partial_destination_copies_only_the_missing_remainder() {
let (src, dst) = pair(true);
// Destination already holds serde 1.0.0 identically (a prior partial run).
dst.put(&id("serde", "1.0.0"), b"serde-bytes").unwrap();
let r = replicate_pairs(
&one_pair(src.clone(), dst.clone()),
&ReplicationOptions::executing(),
);
assert_eq!(r.skipped(), 1, "the already-present artifact is skipped");
assert_eq!(r.copied(), 2, "only the two missing artifacts copy");
// Full mirror regardless of the partial starting point.
for entry in src.list(None, 0).unwrap() {
assert!(
dst.fetch(&entry.id).unwrap().is_some(),
"dest complete after replicate"
);
}
}
#[test]
fn limit_caps_examined_artifacts_and_flags_truncation() {
let (src, dst) = pair(true);
let r = replicate_pairs(
&one_pair(src, dst),
&ReplicationOptions {
execute: true,
limit: 2,
},
);
assert_eq!(r.examined(), 2, "only two artifacts examined under the cap");
assert!(r.truncated, "more source artifacts remained ⇒ truncated");
}
/// A backend that lists an artifact it cannot then produce is a REPORTED
/// failure, never a silent omission: an operator reading the report must be
/// able to tell "3 of 3 copied" from "3 listed, 1 unreadable". RED-when-broken
/// if an unreadable source artifact is ever quietly dropped from the rows.
#[test]
fn an_artifact_the_source_cannot_produce_is_reported_not_dropped() {
/// Lists two artifacts; `fetch` answers for one and refuses the other.
struct HalfBrokenSource;
#[async_trait]
impl RepositoryBackendTrait for HalfBrokenSource {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"half-broken"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
match id.name.as_str() {
"good" => Ok(Some(traits::Bytes::from_static(b"good-bytes"))),
"vanished" => Ok(None), // listed, but there are no bytes
_ => anyhow::bail!("archive entry unreadable"),
}
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
Ok(["good", "vanished", "unreadable"]
.into_iter()
.map(|n| ArtifactEntry {
id: id(n, "1.0.0"),
size_bytes: 10,
content_type: "application/octet-stream".into(),
// A test double knows neither, and says so.
..ArtifactEntry::UNKNOWN_PROVENANCE
})
.collect())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
let dst = MemRepo::new("rust-mirror", true).arc();
let pairs = one_pair(Arc::new(HalfBrokenSource), dst.clone());
let r = replicate_pairs(&pairs, &ReplicationOptions::executing());
assert_eq!(
r.examined(),
3,
"every LISTED artifact yields exactly one row"
);
assert_eq!(r.copied(), 1, "only the readable one is copied");
assert_eq!(r.failed(), 2, "the other two are reported as failures");
assert!(!r.is_complete(), "a report with failures is not complete");
for row in r.items.iter().filter(|i| i.outcome == Outcome::Failed) {
assert!(
!row.detail.is_empty(),
"a failure always carries its reason: {row:?}"
);
assert!(
row.content_id.is_empty(),
"an unread artifact has no content id"
);
}
assert_eq!(
dst.list(None, 0).unwrap().len(),
1,
"only the copied artifact landed"
);
}
/// A source that cannot even be listed contributes no rows — the run reports
/// nothing for that pair rather than inventing outcomes — and the OTHER pairs
/// in the same run still replicate.
#[test]
fn a_source_that_cannot_be_listed_contributes_no_rows_and_spares_its_siblings() {
struct UnlistableSource;
#[async_trait]
impl RepositoryBackendTrait for UnlistableSource {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"unlistable"
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
anyhow::bail!("source archive unreadable")
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
let (src, dst) = pair(true);
let broken_dest = MemRepo::new("broken-mirror", true).arc();
let pairs = vec![
RepoPair {
source_name: "unlistable".into(),
source: Arc::new(UnlistableSource),
dest_name: "broken-mirror".into(),
dest: broken_dest.clone(),
},
RepoPair {
source_name: "rust-dev".into(),
source: src,
dest_name: "rust-mirror".into(),
dest: dst.clone(),
},
];
let r = replicate_pairs(&pairs, &ReplicationOptions::executing());
assert_eq!(
r.examined(),
3,
"only the healthy pair's artifacts are examined"
);
assert!(
r.items.iter().all(|i| i.source_repo == "rust-dev"),
"the unlistable pair contributed no rows at all"
);
assert_eq!(r.copied(), 3, "the healthy pair replicated in full");
assert!(
broken_dest.list(None, 0).unwrap().is_empty(),
"nothing was written for it"
);
}
/// **The source is never written to.** Replication reads the source and writes
/// the destination; a run against a source that would PANIC on `put` proves
/// the direction is structural, not a convention. The source's bytes are
/// byte-identical afterwards, and the copy is byte-identical to them.
#[test]
fn the_source_is_only_ever_read_and_the_copy_is_byte_identical() {
struct WriteProofSource(MemRepo);
#[async_trait]
impl RepositoryBackendTrait for WriteProofSource {
fn caps(&self) -> traits::Caps {
traits::Caps::LISTING
}
fn name(&self) -> &str {
self.0.name()
}
fn format(&self) -> ArtifactFormat {
self.0.format()
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
self.0.fetch(id)
}
fn put(&self, _id: &ArtifactId, _d: &[u8]) -> anyhow::Result<()> {
panic!("replication must NEVER write to the source");
}
fn list(&self, f: Option<&str>, l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
self.0.list(f, l)
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
let payload: Vec<u8> = (0u16..=511).map(|b| (b % 256) as u8).collect();
let src = Arc::new(WriteProofSource(MemRepo::seeded(
"rust-dev",
vec![(id("binary", "1.0.0"), payload.clone())],
)));
let dst = MemRepo::new("rust-mirror", true).arc();
let r = replicate_pairs(
&one_pair(src.clone(), dst.clone()),
&ReplicationOptions::executing(),
);
assert_eq!(r.copied(), 1);
assert_eq!(
dst.fetch(&id("binary", "1.0.0")).unwrap().as_deref(),
Some(payload.as_slice()),
"the copy is byte-identical, binary and all"
);
assert_eq!(
src.fetch(&id("binary", "1.0.0")).unwrap().as_deref(),
Some(payload.as_slice()),
"and the source is untouched"
);
assert_eq!(
r.items[0].content_id,
content_id(&payload),
"the row names the content id"
);
assert_eq!(r.copied_bytes(), payload.len() as i64);
}
}
#[cfg(test)]
mod estate_coverage_tests {
use super::*;
/// **Everything a live holger writes under its data root, enumerated from
/// the code that builds each path** (`server/lib/src/lib.rs`,
/// `jobs.rs`, `metered.rs`, `secure_update.rs`), 2026-09-21.
///
/// The list is deliberately DATA and not a comment: when the server grows a
/// store, this array grows with it and the assertion below says whether the
/// twin was told.
const WHAT_A_PRIMARY_WRITES: &[&str] = &[
"holger-attestations",
"holger-audit",
"holger-props",
"holger-sboms",
"holger-prod-store",
"holger-update-state",
"holger-import-receipts",
"plugins",
"holger-passkeys",
"holger-entitlements",
"meters",
"holger-jobs",
"jobs",
// `purge::RECORDS_DIR` (2026-09-29): a purge the twin never hears of
// survives on the twin for ever — its mirror never deletes.
crate::purge::RECORDS_DIR,
// What a sale leaves (lane P4): `fleet_box::SIGNUPS_REL`, `sold::SOLD_REPOS_REL`.
"fleet/signups.json",
crate::sold::SOLD_REPOS_REL,
// The sync state (lane OX-WGDS2, `sync::STATE_DIR`): the downstreams the console
// enrolled, their revocations, the ledger and the sync key the downstreams trust. A
// main rebuilt from a twin without it has no downstream and signs under a new key.
"holger-sync",
];
fn scratch(tag: &str) -> PathBuf {
let p = std::env::temp_dir().join(format!(
"holger-estate-cover-{tag}-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
));
std::fs::create_dir_all(&p).expect("scratch");
p
}
/// **Every store the primary writes is a store the twin is told about.**
///
/// This is the owner's order read literally — *"the twin has config +
/// data"*. Before this test, `ESTATE` named eight trees and the config, and
/// the primary wrote six more: the console's passkeys (so a promoted twin
/// had NO ONE who could log in), the tenant ledger with its single-use
/// token digests, the entitlement store (gunnar's exact scar), the meter
/// readings, and both job stores.
#[test]
fn every_store_the_primary_writes_is_one_the_twin_is_told_about() {
let named: std::collections::HashSet<&str> = ESTATE.iter().map(|i| i.path).collect();
let missing: Vec<&str> = WHAT_A_PRIMARY_WRITES
.iter()
.copied()
.filter(|p| !named.contains(p))
.collect();
assert!(
missing.is_empty(),
"the primary writes these under its data root and ESTATE does not name them, so a \
twin would never hold them: {missing:?}"
);
// Every entry carries a reason — a bare path in a report tells an
// operator nothing about what a promotion would be missing.
for item in ESTATE {
assert!(!item.what.trim().is_empty(), "{} has no reason", item.path);
}
}
/// **A twin that copied the evidence and not the artifacts says so.**
///
/// The biggest occupant of holger.rs's data root is `store/` — the znippy
/// artifact store, named by the RON config and therefore uncoverable by any
/// fixed `ESTATE` entry. `replicate_estate` moves none of it, and until
/// this run reported `unreplicated` it returned `is_complete() == true` over
/// a twin holding no artifacts at all.
#[test]
fn the_artifact_store_is_named_as_something_this_pass_does_not_move() {
let root = scratch("unreplicated");
let a = root.join("primary");
std::fs::create_dir_all(a.join("holger-props")).unwrap();
// What the emitted holger.ron on holger.rs actually declares:
// `(ron_name: "store", ron_path: "/var/lib/holger/store")`.
std::fs::create_dir_all(a.join("store")).unwrap();
std::fs::write(a.join("store").join("crates.znippy"), b"artifact bytes").unwrap();
// Something nobody has taught the twin about yet.
std::fs::create_dir_all(a.join("holger-something-new")).unwrap();
// A running import's scratch, which a twin must NOT hold.
std::fs::write(a.join(".holger-import.lock"), b"").unwrap();
let report = replicate_estate(&a, &root.join("twin"), &ReplicationOptions::executing())
.expect("the estate replicates");
assert_eq!(
report.unreplicated,
vec!["holger-something-new".to_string(), "store".to_string()],
"both the artifact store and the unknown store are named, the lock file is not"
);
let warning = report.unreplicated_warning().expect("a named warning");
assert!(
warning.contains(holger_errcode::TWIN_ESTATE_INCOMPLETE.code),
"{warning}"
);
assert!(warning.contains("store"), "{warning}");
// A data root with nothing outside ESTATE warns about nothing.
let clean = root.join("clean");
std::fs::create_dir_all(clean.join("holger-props")).unwrap();
let quiet = replicate_estate(
&clean,
&root.join("twin2"),
&ReplicationOptions::executing(),
)
.expect("runs");
assert!(quiet.unreplicated.is_empty());
assert!(quiet.unreplicated_warning().is_none());
let _ = std::fs::remove_dir_all(&root);
}
/// ★ **The audit index is not replicated** (lane HOL-AUDIT): it is derived from the
/// chain, rewritten and merged in place, and every box rebuilds its own; the chain is.
#[test]
fn the_audit_index_is_left_to_each_box_and_the_chain_is_replicated() {
let root = scratch("audit-index");
let a = root.join("primary");
let idx = a
.join("holger-audit")
.join(crate::audit::arrow_ipc::INDEX_DIR);
std::fs::create_dir_all(&idx).unwrap();
std::fs::write(
a.join("holger-audit").join(crate::audit::AUDIT_FILE),
b"{\"seq\":0}\n",
)
.unwrap();
std::fs::write(idx.join(crate::audit::arrow_ipc::MANIFEST), b"{}").unwrap();
std::fs::write(idx.join("MANIFEST.json.tmp"), b"half").unwrap();
let twin = root.join("twin");
let report = replicate_estate(&a, &twin, &ReplicationOptions::executing())
.expect("the estate replicates");
assert!(
twin.join("holger-audit")
.join(crate::audit::AUDIT_FILE)
.is_file(),
"the chain is the twin's too"
);
assert!(
!twin
.join("holger-audit")
.join(crate::audit::arrow_ipc::INDEX_DIR)
.exists(),
"the index is the twin's own to build"
);
assert!(
report.items.iter().all(|i| i.failed == 0),
"{:?}",
report.items.iter().map(|i| &i.failures).collect::<Vec<_>>()
);
let _ = std::fs::remove_dir_all(&root);
}
}