//! Live artifact-lineage view-model — the **proxy-cache-filling** facet.
//!
//! holger is a pull-through proxy/cache. As it caches and serves artifacts, this
//! view-model maintains a live **lineage graph** of what the cache holds:
//!
//! * **nodes** — every cached artifact (grouped/clustered by repository +
//! [`Ecosystem`]), plus the repository that requested it and the upstream it was
//! proxied from,
//! * **edges** — the lineage: artifact → the repo that requested it
//! ([`EdgeKind::RequestedBy`]) and artifact → its upstream/proxied source
//! ([`EdgeKind::ProxiedFrom`]),
//! * **events** — a time-ordered list of arrivals (the "when did the cache fill"
//! temporal view the helix winds along a coil).
//!
//! The feed is **transport-agnostic**: the model consumes a plain
//! [`LineageEvent`] and knows nothing about how it was produced. Two producers are
//! supported:
//!
//! * **live** — [`LineageView::connect`] attaches an `mpsc::Receiver<LineageEvent>`;
//! each [`LineageView::pump`] drains newly-arrived events and grows the graph
//! incrementally (nodes pop in as the cache fills),
//! * **replay** — [`LineageView::replay`] folds an iterator of events (e.g. the
//! whole history) into the same state.
//!
//! The seam onto holger's real data feed — the append-only Arrow-IPC **audit log**
//! (`server_lib::audit`) — lives in the gated [`audit_tap`] submodule: it maps an
//! `AuditEvent` → [`LineageEvent`], provides [`replay_audit_dir`](LineageView::replay_audit_dir)
//! over a segment directory, and a [`ChannelAuditLog`] tee that a running holger can
//! install as its `Arc<dyn AuditLog>` so every fetch/put streams straight into a
//! connected `LineageView`.
//!
//! There is **no egui here**: like every view in [`crate::data`], the whole state
//! is observable as [`state_json`](LineageView::state_json) for headless robot tests.
use std::collections::{HashMap, HashSet};
use std::sync::mpsc::{Receiver, Sender};
use serde::Serialize;
use serde_json::Value;
/// The package ecosystem an artifact belongs to — the cluster key the 3D fill
/// groups nodes by and the lane the helix winds each arrival onto. Derived from
/// the repository name (and, weakly, the artifact shape) since holger's audit
/// record carries no explicit format column.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Ecosystem {
Maven,
Pypi,
Rust,
Npm,
Nuget,
Go,
Gem,
Deb,
Rpm,
Helm,
Docker,
Conda,
Composer,
/// Unclassified / not one of the known package ecosystems.
Other,
}
impl Ecosystem {
/// Classify a repository (and artifact) into an ecosystem by substring match on
/// the repo name — the ecosystem is not an audit column, so it is recovered
/// from the repo the request targeted (the same repo→backend mapping holger
/// wires per `znippy-{maven,python,rust,…}` crate), falling back to the artifact
/// shape and finally [`Ecosystem::Other`].
pub fn classify(repo: &str, artifact: &str) -> Ecosystem {
let r = repo.to_ascii_lowercase();
let has = |needles: &[&str]| needles.iter().any(|n| r.contains(n));
if has(&["maven", "mvn", "jar", "gradle"]) {
Ecosystem::Maven
} else if has(&["pypi", "python", "wheel", "pip"]) {
Ecosystem::Pypi
} else if has(&["rust", "crate", "cargo"]) {
Ecosystem::Rust
} else if has(&["npm", "node"]) {
Ecosystem::Npm
} else if has(&["nuget", "dotnet"]) {
Ecosystem::Nuget
} else if has(&["golang", "goproxy", "go-"]) || r == "go" {
Ecosystem::Go
} else if has(&["gem", "ruby"]) {
Ecosystem::Gem
} else if has(&["deb", "apt", "ubuntu", "debian"]) {
Ecosystem::Deb
} else if has(&["rpm", "yum", "dnf", "rocky", "fedora"]) {
Ecosystem::Rpm
} else if has(&["helm", "chart"]) {
Ecosystem::Helm
} else if has(&["docker", "oci", "image", "registry"]) {
Ecosystem::Docker
} else if has(&["conda", "anaconda"]) {
Ecosystem::Conda
} else if has(&["composer", "php", "packagist"]) {
Ecosystem::Composer
} else if artifact.contains(".crate") {
Ecosystem::Rust
} else if artifact.ends_with(".whl") || artifact.ends_with(".tar.gz") {
Ecosystem::Pypi
} else if artifact.ends_with(".jar") {
Ecosystem::Maven
} else {
Ecosystem::Other
}
}
/// Stable lowercase label — the helix lane name and the cluster key surfaced in
/// `state_json`.
pub fn label(self) -> &'static str {
match self {
Ecosystem::Maven => "maven",
Ecosystem::Pypi => "pypi",
Ecosystem::Rust => "rust",
Ecosystem::Npm => "npm",
Ecosystem::Nuget => "nuget",
Ecosystem::Go => "go",
Ecosystem::Gem => "gem",
Ecosystem::Deb => "deb",
Ecosystem::Rpm => "rpm",
Ecosystem::Helm => "helm",
Ecosystem::Docker => "docker",
Ecosystem::Conda => "conda",
Ecosystem::Composer => "composer",
Ecosystem::Other => "other",
}
}
/// The public upstream host a pull-through proxy of this ecosystem fills from.
/// Used to synthesize the `proxied-from` source when the audit record does not
/// carry an explicit upstream (holger's audit log records the local
/// fetch/put, not the origin — see the module docs). `None` for ecosystems with
/// no single canonical public registry.
pub fn upstream_host(self) -> Option<&'static str> {
Some(match self {
Ecosystem::Maven => "repo1.maven.org",
Ecosystem::Pypi => "pypi.org",
Ecosystem::Rust => "crates.io",
Ecosystem::Npm => "registry.npmjs.org",
Ecosystem::Nuget => "nuget.org",
Ecosystem::Go => "proxy.golang.org",
Ecosystem::Gem => "rubygems.org",
Ecosystem::Conda => "conda.anaconda.org",
Ecosystem::Composer => "packagist.org",
Ecosystem::Deb | Ecosystem::Rpm | Ecosystem::Helm | Ecosystem::Docker | Ecosystem::Other => {
return None
}
})
}
/// A neon RGB base colour for this ecosystem cluster (the egui-free palette the
/// `gui` layer lifts into a `Color32`).
pub fn color(self) -> [u8; 3] {
match self {
Ecosystem::Maven => [224, 122, 95],
Ecosystem::Pypi => [75, 150, 220],
Ecosystem::Rust => [222, 165, 132],
Ecosystem::Npm => [203, 86, 85],
Ecosystem::Nuget => [120, 100, 210],
Ecosystem::Go => [80, 200, 220],
Ecosystem::Gem => [220, 70, 70],
Ecosystem::Deb => [200, 70, 110],
Ecosystem::Rpm => [90, 130, 200],
Ecosystem::Helm => [70, 170, 200],
Ecosystem::Docker => [70, 150, 225],
Ecosystem::Conda => [90, 180, 165],
Ecosystem::Composer => [150, 165, 205],
Ecosystem::Other => [160, 160, 172],
}
}
}
/// The operation an arrival describes — a transport-agnostic mirror of holger's
/// `audit::AuditAction`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LineageAction {
Download,
Upload,
Delete,
List,
Promote,
Other,
}
impl LineageAction {
/// Stable lowercase wire string.
pub fn as_str(self) -> &'static str {
match self {
LineageAction::Download => "download",
LineageAction::Upload => "upload",
LineageAction::Delete => "delete",
LineageAction::List => "list",
LineageAction::Promote => "promote",
LineageAction::Other => "other",
}
}
}
/// One cache-fill event the [`LineageView`] consumes — the transport-agnostic input
/// the model maps into a node + edges + a time event. It carries everything an
/// `audit::AuditEvent` does, plus an optional explicit `upstream` (a real feed can
/// supply the proxied origin; when `None` the model derives one from the
/// ecosystem).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageEvent {
/// Event time, nanoseconds since the Unix epoch (UTC).
pub ts_nanos: i64,
/// Authenticated principal (or `"anonymous"`).
pub ident: String,
/// The operation performed.
pub action: LineageAction,
/// Repository the request targeted (the cluster + `requested-by` node).
pub repo: String,
/// Artifact identifier (`namespace/name@version` or an archive path).
pub artifact: String,
/// Client source address (`ip:port`), if the transport exposed one.
pub source_ip: String,
/// Result status (HTTP-like: 200/403/404/…).
pub status: u16,
/// Bytes transferred.
pub bytes: u64,
/// Explicit proxied-from origin, when a real feed knows it; else the model
/// derives one from the ecosystem's canonical upstream.
pub upstream: Option<String>,
}
impl LineageEvent {
/// Build an arrival with no explicit upstream (the common case — the audit log
/// records the local fetch/put, and the origin is derived from the ecosystem).
#[allow(clippy::too_many_arguments)]
pub fn new(
ts_nanos: i64,
ident: impl Into<String>,
action: LineageAction,
repo: impl Into<String>,
artifact: impl Into<String>,
source_ip: impl Into<String>,
status: u16,
bytes: u64,
) -> Self {
Self {
ts_nanos,
ident: ident.into(),
action,
repo: repo.into(),
artifact: artifact.into(),
source_ip: source_ip.into(),
status,
bytes,
upstream: None,
}
}
/// Attach an explicit proxied-from origin (builder style).
pub fn with_upstream(mut self, upstream: impl Into<String>) -> Self {
self.upstream = Some(upstream.into());
self
}
}
/// What a lineage-graph node represents.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum NodeKind {
/// A cached artifact.
Artifact,
/// A repository that requested artifacts (the `requested-by` anchor).
Repo,
/// An upstream / proxied-from source.
Upstream,
}
/// The lineage an edge encodes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EdgeKind {
/// artifact → the repository that requested it.
RequestedBy,
/// artifact → its upstream / proxied source.
ProxiedFrom,
}
/// One node in the live lineage graph.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageNode {
/// Stable id (`art:{repo}/{artifact}`, `repo:{repo}`, or `up:{host}`).
pub id: String,
/// Display label.
pub label: String,
/// What the node is.
pub kind: NodeKind,
/// The ecosystem cluster this node belongs to.
pub ecosystem: Ecosystem,
/// The repository the node is associated with (its own name for a repo node).
pub repo: String,
/// When the artifact first landed in the cache (nanoseconds UTC); the arrival
/// time for repo/upstream nodes.
pub cached_at_nanos: i64,
/// Artifact size in bytes (max seen); 0 for repo/upstream nodes.
pub size: u64,
/// The proxied-from upstream, if known.
pub upstream: Option<String>,
/// How many times this node has been touched (cache hits + the initial fill).
pub hits: u32,
}
/// One directed lineage edge (by node id, so it survives incremental growth).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageEdge {
pub from: String,
pub to: String,
pub kind: EdgeKind,
}
/// One time-ordered arrival — the temporal event the helix winds onto a per-
/// ecosystem coil.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct TimeEvent {
pub ts_nanos: i64,
pub artifact: String,
pub ecosystem: Ecosystem,
pub repo: String,
pub action: LineageAction,
}
/// The live lineage graph state — nodes (deduplicated), edges (deduplicated), and a
/// time-ordered event list, grown by [`ingest`](Self::ingest) from either a live
/// channel ([`connect`](Self::connect) + [`pump`](Self::pump)) or a replay
/// ([`replay`](Self::replay)). Every field is observable via
/// [`state_json`](Self::state_json).
#[derive(Default, Serialize)]
pub struct LineageView {
/// Cached-artifact / repository / upstream nodes, in first-seen order (so a
/// renderer sees them "pop in" as the cache fills).
pub nodes: Vec<LineageNode>,
/// Lineage edges by node id.
pub edges: Vec<LineageEdge>,
/// Time-ordered arrivals.
pub events: Vec<TimeEvent>,
/// Total arrivals ingested (including cache hits on already-known artifacts).
pub total_events: u64,
/// Total bytes transferred across all arrivals.
pub total_bytes: u64,
/// id → index into `nodes`, for O(1) dedup/upsert. Not observable.
#[serde(skip)]
node_index: HashMap<String, usize>,
/// Seen `(from,to)` pairs, for edge dedup. Not observable.
#[serde(skip)]
edge_seen: HashSet<(String, String)>,
/// The live feed, when connected. Drained by [`pump`](Self::pump).
#[serde(skip)]
rx: Option<Receiver<LineageEvent>>,
}
/// Convenience: a fresh unbounded lineage feed channel. The `Sender` half is
/// handed to a producer (e.g. a [`ChannelAuditLog`] tee installed on a running
/// holger, or a demo driver); the `Receiver` is handed to
/// [`LineageView::connect`].
pub fn channel() -> (Sender<LineageEvent>, Receiver<LineageEvent>) {
std::sync::mpsc::channel()
}
impl LineageView {
/// A fresh, empty lineage view with no feed attached.
pub fn new() -> Self {
Self::default()
}
/// Attach a live feed. Subsequent [`pump`](Self::pump) calls drain it.
pub fn connect(&mut self, rx: Receiver<LineageEvent>) {
self.rx = Some(rx);
}
/// `true` once a live feed is attached.
pub fn is_live(&self) -> bool {
self.rx.is_some()
}
/// Drain all currently-available events from the live feed and fold each into
/// the graph. Returns how many were ingested. Non-blocking; safe to call every
/// frame. A disconnected sender is tolerated (the feed simply goes quiet).
pub fn pump(&mut self) -> usize {
let drained: Vec<LineageEvent> = match &self.rx {
Some(rx) => rx.try_iter().collect(),
None => return 0,
};
let n = drained.len();
for ev in drained {
self.ingest(ev);
}
n
}
/// Fold a batch (e.g. the whole audit history) into the graph — the
/// replay-from-log mode.
pub fn replay(&mut self, events: impl IntoIterator<Item = LineageEvent>) {
for ev in events {
self.ingest(ev);
}
}
/// Number of distinct cached **artifact** nodes (excludes repo/upstream nodes).
pub fn artifact_count(&self) -> usize {
self.nodes.iter().filter(|n| n.kind == NodeKind::Artifact).count()
}
/// Fold one arrival into the live graph: upsert the artifact node (clustered by
/// ecosystem), the requesting repo node and the proxied-from upstream node, wire
/// the two lineage edges, and append the time event. An arrival with no specific
/// artifact (a bare listing) is counted but adds no node.
pub fn ingest(&mut self, ev: LineageEvent) {
self.total_events += 1;
self.total_bytes = self.total_bytes.saturating_add(ev.bytes);
if ev.artifact.is_empty() {
return; // a bare list/other with no artifact — nothing to cache.
}
let eco = Ecosystem::classify(&ev.repo, &ev.artifact);
let upstream = ev
.upstream
.clone()
.or_else(|| eco.upstream_host().map(|h| h.to_string()));
// The repository node (requested-by anchor).
let repo_id = format!("repo:{}", ev.repo);
self.upsert(&repo_id, || LineageNode {
id: repo_id.clone(),
label: ev.repo.clone(),
kind: NodeKind::Repo,
ecosystem: eco,
repo: ev.repo.clone(),
cached_at_nanos: ev.ts_nanos,
size: 0,
upstream: None,
hits: 0,
});
// The cached artifact node. `upsert` owns the hit count (1 on the first
// fill, +1 per later cache hit); here we only refresh size/upstream from the
// latest arrival.
let art_id = format!("art:{}/{}", ev.repo, ev.artifact);
self.upsert(&art_id, || LineageNode {
id: art_id.clone(),
label: ev.artifact.clone(),
kind: NodeKind::Artifact,
ecosystem: eco,
repo: ev.repo.clone(),
cached_at_nanos: ev.ts_nanos,
size: ev.bytes,
upstream: upstream.clone(),
hits: 0,
});
if let Some(&i) = self.node_index.get(&art_id) {
let n = &mut self.nodes[i];
n.size = n.size.max(ev.bytes);
if n.upstream.is_none() {
n.upstream = upstream.clone();
}
}
// artifact → repo (requested-by).
self.add_edge(&art_id, &repo_id, EdgeKind::RequestedBy);
// The upstream / proxied-from source node + edge.
if let Some(host) = &upstream {
let up_id = format!("up:{host}");
self.upsert(&up_id, || LineageNode {
id: up_id.clone(),
label: host.clone(),
kind: NodeKind::Upstream,
ecosystem: eco,
repo: ev.repo.clone(),
cached_at_nanos: ev.ts_nanos,
size: 0,
upstream: None,
hits: 0,
});
self.add_edge(&art_id, &up_id, EdgeKind::ProxiedFrom);
}
self.events.push(TimeEvent {
ts_nanos: ev.ts_nanos,
artifact: ev.artifact,
ecosystem: eco,
repo: ev.repo,
action: ev.action,
});
}
/// Insert `id`'s node (from `make`) if unseen; always bump its `hits`. Returns
/// `true` if a new node was created.
fn upsert(&mut self, id: &str, make: impl FnOnce() -> LineageNode) -> bool {
if let Some(&i) = self.node_index.get(id) {
self.nodes[i].hits = self.nodes[i].hits.saturating_add(1);
false
} else {
let idx = self.nodes.len();
let mut node = make();
node.hits = 1;
self.nodes.push(node);
self.node_index.insert(id.to_string(), idx);
true
}
}
/// Add a deduplicated directed edge.
fn add_edge(&mut self, from: &str, to: &str, kind: EdgeKind) {
let key = (from.to_string(), to.to_string());
if self.edge_seen.insert(key) {
self.edges.push(LineageEdge {
from: from.to_string(),
to: to.to_string(),
kind,
});
}
}
/// The whole lineage state as observable JSON (the headless robot contract).
pub fn state_json(&self) -> Value {
serde_json::to_value(self).unwrap_or(Value::Null)
}
}
/// A self-contained demo backlog: a realistic multi-ecosystem cache-fill sequence
/// (rust/pypi/maven/npm/go arrivals with a couple of repeat hits) the `gui` layer
/// drips one-per-frame through the LIVE channel so the graph visibly fills and
/// rotates when no real audit feed is wired. Also handy as a test fixture.
pub fn demo_events() -> Vec<LineageEvent> {
const S: i64 = 1_000_000_000; // one second in nanos
let base = 1_700_000_000 * S;
let mk = |i: i64, action, repo: &str, artifact: &str, bytes| {
LineageEvent::new(base + i * S, "anonymous", action, repo, artifact, "10.0.0.7:5555", 200, bytes)
};
use LineageAction::{Download, Upload};
vec![
mk(0, Download, "rust-proxy", "serde@1.0.203", 92_160),
mk(1, Download, "pypi-proxy", "numpy-1.26.4.whl", 5_242_880),
mk(2, Download, "rust-proxy", "tokio@1.37.0", 720_896),
mk(3, Download, "maven-central", "org.slf4j/slf4j-api@2.0.12", 61_440),
mk(4, Download, "npm-proxy", "react@18.2.0", 311_296),
mk(5, Download, "rust-proxy", "serde@1.0.203", 92_160), // cache hit
mk(6, Download, "pypi-proxy", "requests-2.31.0.whl", 143_360),
mk(7, Download, "go-proxy", "golang.org/x/text@v0.14.0", 1_048_576),
mk(8, Download, "maven-central", "com.google.guava/guava@33.1.0", 2_883_584),
mk(9, Upload, "rust-dev", "holger-server-ui@0.1.2", 204_800),
mk(10, Download, "npm-proxy", "lodash@4.17.21", 552_960),
mk(11, Download, "rust-proxy", "tokio@1.37.0", 720_896), // cache hit
mk(12, Download, "pypi-proxy", "pandas-2.2.1.whl", 12_582_912),
mk(13, Download, "maven-central", "org.junit/junit-jupiter@5.10.2", 122_880),
mk(14, Download, "go-proxy", "github.com/gin-gonic/gin@v1.9.1", 819_200),
mk(15, Download, "rust-proxy", "anyhow@1.0.81", 45_056),
]
}
// ── audit-log seam ──────────────────────────────────────────────────────────
//
// The adapter onto holger's real feed needs `server_lib::audit`, which is only a
// dependency under `gui` (real build) or as a dev-dependency (`test`). Gated with
// the same `any(feature = "gui", test)` idiom the runner methods in `crate::data`
// use, so the transport-agnostic model above compiles with no server-lib in the
// default build.
#[cfg(any(feature = "gui-core", test))]
mod audit_tap {
use super::*;
use server_lib::audit::{AuditAction, AuditEvent, AuditError, AuditLog, ArrowIpcAuditLog};
use std::path::Path;
impl From<AuditAction> for LineageAction {
fn from(a: AuditAction) -> Self {
match a {
AuditAction::Download => LineageAction::Download,
AuditAction::Upload => LineageAction::Upload,
AuditAction::Delete => LineageAction::Delete,
AuditAction::List => LineageAction::List,
AuditAction::Promote => LineageAction::Promote,
AuditAction::Other => LineageAction::Other,
}
}
}
impl From<&AuditEvent> for LineageEvent {
/// Map one audit record onto a lineage event. Ecosystem + upstream are not
/// audit columns, so they are derived downstream in [`LineageView::ingest`];
/// `upstream` is left `None` here (the audit log records the local
/// fetch/put, not the proxied origin).
fn from(e: &AuditEvent) -> Self {
LineageEvent {
ts_nanos: e.ts_nanos,
ident: e.ident.clone(),
action: e.action.into(),
repo: e.repo.clone(),
artifact: e.artifact.clone(),
source_ip: e.source_ip.clone(),
status: e.status,
bytes: e.bytes,
upstream: None,
}
}
}
impl LineageView {
/// Replay-from-log mode over holger's audit stream: read every
/// append-only Arrow-IPC segment under `dir` (oldest-first) and fold each
/// recorded fetch/put into the lineage graph. Returns the number of events
/// replayed. A missing directory replays as empty.
pub fn replay_audit_dir(&mut self, dir: impl AsRef<Path>) -> anyhow::Result<usize> {
let events = ArrowIpcAuditLog::read_dir(dir)
.map_err(|e: AuditError| anyhow::anyhow!("audit replay failed: {e}"))?;
let n = events.len();
for ae in &events {
self.ingest(LineageEvent::from(ae));
}
Ok(n)
}
}
/// An [`AuditLog`] that **tees** every recorded event into a lineage feed
/// channel — the live tap. Install it (or a fan-out wrapper) as a running
/// holger's `Arc<dyn AuditLog>` and hand the paired `Receiver` to
/// [`LineageView::connect`]; every fetch/put then streams live into the graph.
/// `record` never fails (a disconnected receiver is dropped silently), matching
/// holger's fire-and-forget audit contract.
pub struct ChannelAuditLog {
tx: Sender<LineageEvent>,
}
impl ChannelAuditLog {
/// Wrap a lineage-feed sender (the `Sender` half of [`super::channel`]).
pub fn new(tx: Sender<LineageEvent>) -> Self {
Self { tx }
}
}
impl AuditLog for ChannelAuditLog {
fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
let _ = self.tx.send(LineageEvent::from(&event));
Ok(())
}
}
}
#[cfg(any(feature = "gui-core", test))]
pub use audit_tap::ChannelAuditLog;
#[cfg(test)]
mod tests {
use super::*;
use server_lib::audit::{AuditAction, AuditEvent, AuditLog};
/// Emit one functional-status row into the `nornir test` matrix. Gated behind
/// `--features testmatrix` so a release build strips it entirely.
#[cfg(feature = "testmatrix")]
fn fstatus(component: &str, check: &str, ok: bool, detail: &str) {
nornir_testmatrix::functional_status(component, check, ok, detail);
}
/// An audit `AuditEvent` maps → `LineageEvent`, and `ingest` derives the
/// ecosystem cluster (from the repo) and the proxied-from upstream (from the
/// ecosystem) even though neither is an audit column.
#[test]
fn audit_event_maps_and_derives_ecosystem_and_upstream() {
let ae = AuditEvent::new(
"anonymous",
AuditAction::Download,
"rust-proxy",
"serde@1.0.203",
"10.0.0.9:4444",
200,
92_160,
);
let mut view = LineageView::new();
view.ingest(LineageEvent::from(&ae));
let art = view
.nodes
.iter()
.find(|n| n.kind == NodeKind::Artifact)
.expect("artifact node");
let eco_ok = art.ecosystem == Ecosystem::Rust;
let up_ok = art.upstream.as_deref() == Some("crates.io");
assert!(eco_ok, "repo 'rust-proxy' derives the rust ecosystem: {:?}", art.ecosystem);
assert!(up_ok, "rust upstream derived as crates.io: {:?}", art.upstream);
// artifact → repo (requested-by) + artifact → upstream (proxied-from).
assert_eq!(view.edges.len(), 2, "two lineage edges wired");
assert!(view
.edges
.iter()
.any(|e| e.kind == EdgeKind::RequestedBy && e.to == "repo:rust-proxy"));
assert!(view
.edges
.iter()
.any(|e| e.kind == EdgeKind::ProxiedFrom && e.to == "up:crates.io"));
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_audit_map_derive",
eco_ok && up_ok && view.edges.len() == 2,
&format!(
"AuditEvent→LineageEvent: eco={} upstream={:?} edges={}",
art.ecosystem.label(),
art.upstream,
view.edges.len()
),
);
}
/// The `ChannelAuditLog` tap: install it as a holger `Arc<dyn AuditLog>`, record
/// events through the SAME `record()` the gRPC fetch/put path calls, and prove
/// they stream live into a connected `LineageView` via `pump()`.
#[test]
fn channel_audit_tap_streams_live_into_view() {
let (tx, rx) = channel();
let mut view = LineageView::new();
view.connect(rx);
assert!(view.is_live());
let sink: std::sync::Arc<dyn AuditLog> = std::sync::Arc::new(ChannelAuditLog::new(tx));
// Two ecosystems + one cache hit (serde downloaded twice).
sink.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:1", 200, 92_160))
.unwrap();
sink.record(AuditEvent::new("anon", AuditAction::Download, "pypi-proxy", "numpy-1.26.4.whl", "ip:2", 200, 5_242_880))
.unwrap();
sink.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:3", 200, 92_160))
.unwrap();
let drained = view.pump();
assert_eq!(drained, 3, "pump drained all three recorded events");
let s = view.state_json();
assert_eq!(s["total_events"].as_u64(), Some(3), "state_json event tally: {s}");
// Two distinct artifacts (serde deduped across its two hits) → two artifact nodes.
assert_eq!(view.artifact_count(), 2, "serde deduped across two hits");
// The serde artifact recorded a cache hit (>1 touch).
let serde = view
.nodes
.iter()
.find(|n| n.label == "serde@1.0.203")
.expect("serde artifact node");
assert!(serde.hits >= 2, "serde touched twice (fill + cache hit): {}", serde.hits);
// Grouped by ecosystem: rust + pypi clusters present.
let ecos: Vec<&str> = view
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Artifact)
.map(|n| n.ecosystem.label())
.collect();
assert!(ecos.contains(&"rust") && ecos.contains(&"pypi"), "both clusters: {ecos:?}");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_channel_tap_live",
drained == 3 && view.artifact_count() == 2 && serde.hits >= 2,
&format!(
"ChannelAuditLog tap → pump: {drained} events, {} artifacts, serde hits {}",
view.artifact_count(),
serde.hits
),
);
}
/// Replay-from-log: write real audit events through `ArrowIpcAuditLog` into a
/// segment dir, then `replay_audit_dir` folds them back into a fresh view.
#[test]
fn replay_from_audit_dir_round_trips() {
use server_lib::audit::ArrowIpcAuditLog;
let dir = std::env::temp_dir().join(format!("holger-lineage-replay-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
{
let log = ArrowIpcAuditLog::new(&dir).expect("open audit segment");
log.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:1", 200, 92_160))
.unwrap();
log.record(AuditEvent::new("anon", AuditAction::Upload, "maven-central", "org.slf4j/slf4j-api@2.0.12", "ip:2", 200, 61_440))
.unwrap();
// dropped → IPC stream finished.
}
let mut view = LineageView::new();
let n = view.replay_audit_dir(&dir).expect("replay");
let _ = std::fs::remove_dir_all(&dir);
assert_eq!(n, 2, "replayed both recorded events");
assert_eq!(view.artifact_count(), 2, "two cached artifacts after replay");
assert_eq!(view.events.len(), 2, "two time events after replay");
let ecos: HashSet<&str> = view
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Artifact)
.map(|n| n.ecosystem.label())
.collect();
assert!(ecos.contains(&"rust") && ecos.contains(&"maven"), "replayed clusters: {ecos:?}");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_replay_audit_dir",
n == 2 && view.artifact_count() == 2 && view.events.len() == 2,
&format!("replay_audit_dir: {n} events → {} artifacts", view.artifact_count()),
);
}
/// The demo backlog folds cleanly, dedups repeat hits, and clusters by ecosystem
/// — the fixture the gui layer drips through the live channel.
#[test]
fn demo_events_fill_and_dedup() {
let mut view = LineageView::new();
view.replay(demo_events());
assert_eq!(view.total_events as usize, demo_events().len(), "every demo event counted");
// serde and tokio each appear twice (cache hits) → fewer artifact nodes than events.
assert!(view.artifact_count() < demo_events().len(), "repeat hits deduped");
let clusters: HashSet<&str> = view
.nodes
.iter()
.filter(|n| n.kind == NodeKind::Artifact)
.map(|n| n.ecosystem.label())
.collect();
assert!(clusters.len() >= 4, "multi-ecosystem fill: {clusters:?}");
// state_json exposes the full picture for the robot.
let s = view.state_json();
assert!(s["nodes"].as_array().map(|a| !a.is_empty()).unwrap_or(false));
assert!(s["edges"].as_array().map(|a| !a.is_empty()).unwrap_or(false));
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_demo_fill",
view.artifact_count() < demo_events().len() && clusters.len() >= 4,
&format!("demo fill: {} events → {} artifacts, {} clusters", view.total_events, view.artifact_count(), clusters.len()),
);
}
/// `Ecosystem::classify` recovers the cluster from the repo name across every
/// known ecosystem, falls back to the artifact shape when the repo name is
/// unhelpful, and lands on `Other` for the truly unknown — the grouping key the
/// 3D fill and the helix lanes rely on.
#[test]
fn classify_covers_ecosystems_and_falls_back() {
// Repo-name driven (the primary path — the repo→backend mapping holger wires).
let by_repo = [
("maven-central", Ecosystem::Maven),
("pypi-proxy", Ecosystem::Pypi),
("rust-crates", Ecosystem::Rust),
("npm-proxy", Ecosystem::Npm),
("nuget-feed", Ecosystem::Nuget),
("go-proxy", Ecosystem::Go),
("ruby-gems", Ecosystem::Gem),
("debian-apt", Ecosystem::Deb),
("rpm-yum", Ecosystem::Rpm),
("helm-charts", Ecosystem::Helm),
("docker-registry", Ecosystem::Docker),
("conda-forge", Ecosystem::Conda),
("composer-packagist", Ecosystem::Composer),
];
for (repo, want) in by_repo {
assert_eq!(Ecosystem::classify(repo, "whatever"), want, "repo {repo:?}");
}
// Artifact-shape fallback when the repo name carries no ecosystem hint.
assert_eq!(Ecosystem::classify("mirror", "foo-0.1.0.crate"), Ecosystem::Rust);
assert_eq!(Ecosystem::classify("mirror", "numpy-1.26.4.whl"), Ecosystem::Pypi);
assert_eq!(Ecosystem::classify("mirror", "slf4j-api-2.0.12.jar"), Ecosystem::Maven);
// Truly unknown → Other. Its label round-trips too.
let other = Ecosystem::classify("mirror", "readme.txt");
assert_eq!(other, Ecosystem::Other);
assert_eq!(other.label(), "other");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_classify_matrix",
by_repo.iter().all(|(r, w)| Ecosystem::classify(r, "x") == *w)
&& Ecosystem::classify("mirror", "readme.txt") == Ecosystem::Other,
"classify: 13 ecosystems by repo + artifact-shape fallback + Other",
);
}
/// An explicit `upstream` on the event (a real feed that knows the proxied
/// origin) overrides the ecosystem-derived default — the proxied-from node and
/// edge point at the supplied host, not the canonical public registry.
#[test]
fn explicit_upstream_overrides_derived() {
let mut view = LineageView::new();
// rust would derive crates.io; the feed instead names an internal mirror.
view.ingest(
LineageEvent::new(1, "anon", LineageAction::Download, "rust-proxy", "serde@1.0", "ip:1", 200, 10)
.with_upstream("mirror.corp.internal"),
);
let art = view.nodes.iter().find(|n| n.kind == NodeKind::Artifact).expect("artifact");
assert_eq!(art.upstream.as_deref(), Some("mirror.corp.internal"), "explicit upstream wins");
assert!(
view.edges.iter().any(|e| e.kind == EdgeKind::ProxiedFrom && e.to == "up:mirror.corp.internal"),
"proxied-from edge targets the explicit mirror, not crates.io: {:?}",
view.edges
);
assert!(
!view.edges.iter().any(|e| e.to == "up:crates.io"),
"the derived crates.io upstream is not used when one is supplied",
);
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_explicit_upstream",
art.upstream.as_deref() == Some("mirror.corp.internal"),
"explicit LineageEvent.upstream overrides the ecosystem-derived registry",
);
}
/// An ecosystem with no single canonical public registry (helm/docker/…) derives
/// no upstream: only the requested-by edge is wired, no proxied-from node/edge —
/// the graph never invents a phantom source.
#[test]
fn no_upstream_ecosystem_wires_only_requested_by() {
let mut view = LineageView::new();
view.ingest(LineageEvent::new(1, "anon", LineageAction::Download, "helm-charts", "njord-0.1.0.tgz", "ip:1", 200, 4096));
let art = view.nodes.iter().find(|n| n.kind == NodeKind::Artifact).expect("artifact");
assert_eq!(art.ecosystem, Ecosystem::Helm);
assert!(art.upstream.is_none(), "helm has no canonical upstream host");
assert!(Ecosystem::Helm.upstream_host().is_none());
// Exactly one edge: artifact → repo. No proxied-from, no upstream node.
assert_eq!(view.edges.len(), 1, "only the requested-by edge: {:?}", view.edges);
assert_eq!(view.edges[0].kind, EdgeKind::RequestedBy);
assert!(!view.nodes.iter().any(|n| n.kind == NodeKind::Upstream), "no phantom upstream node");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_no_upstream",
art.upstream.is_none() && view.edges.len() == 1 && !view.nodes.iter().any(|n| n.kind == NodeKind::Upstream),
"helm arrival: requested-by only, no derived proxied-from",
);
}
/// A bare listing (no specific artifact) is tallied into `total_events` /
/// `total_bytes` but adds no node/edge/time-event — a directory `list` doesn't
/// fill the cache.
#[test]
fn bare_list_event_counts_but_adds_no_node() {
let mut view = LineageView::new();
view.ingest(LineageEvent::new(1, "anon", LineageAction::List, "rust-proxy", "", "ip:1", 200, 512));
assert_eq!(view.total_events, 1, "the list is tallied");
assert_eq!(view.total_bytes, 512, "its bytes are tallied");
assert!(view.nodes.is_empty(), "no node for a bare listing");
assert!(view.edges.is_empty(), "no edge for a bare listing");
assert!(view.events.is_empty(), "no time-event for a bare listing");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_bare_list",
view.total_events == 1 && view.nodes.is_empty() && view.events.is_empty(),
"bare list arrival counted but adds no cache node",
);
}
/// The live feed grows the graph incrementally across successive `pump`s (nodes
/// "pop in" frame by frame), and the first-seen node order is preserved — the
/// contract the renderer relies on to animate arrivals landing.
#[test]
fn incremental_pump_pops_nodes_in_across_frames() {
let (tx, rx) = channel();
let mut view = LineageView::new();
view.connect(rx);
// Frame 1: one arrival.
tx.send(LineageEvent::new(1, "a", LineageAction::Download, "rust-proxy", "serde@1.0", "ip:1", 200, 10)).unwrap();
assert_eq!(view.pump(), 1, "first pump drains one");
assert_eq!(view.artifact_count(), 1, "one artifact popped in");
// Frame 2: nothing new yet → pump is a no-op, state unchanged.
assert_eq!(view.pump(), 0, "empty pump");
assert_eq!(view.artifact_count(), 1);
// Frame 3: two more arrivals in a different cluster.
tx.send(LineageEvent::new(2, "a", LineageAction::Download, "pypi-proxy", "numpy.whl", "ip:1", 200, 20)).unwrap();
tx.send(LineageEvent::new(3, "a", LineageAction::Download, "npm-proxy", "react", "ip:1", 200, 30)).unwrap();
assert_eq!(view.pump(), 2, "third pump drains the two");
assert_eq!(view.artifact_count(), 3, "graph grew to three artifacts");
// A dropped sender is tolerated: the feed just goes quiet.
drop(tx);
assert_eq!(view.pump(), 0, "disconnected sender pumps empty, no panic");
// First-seen order preserved: serde's artifact node predates numpy's.
let art_ids: Vec<&str> = view.nodes.iter().filter(|n| n.kind == NodeKind::Artifact).map(|n| n.label.as_str()).collect();
assert_eq!(art_ids, ["serde@1.0", "numpy.whl", "react"], "first-seen pop-in order");
#[cfg(feature = "testmatrix")]
fstatus(
"holger-server-ui",
"lineage_incremental_pump",
view.artifact_count() == 3 && art_ids == ["serde@1.0", "numpy.whl", "react"],
"live feed grows the graph incrementally in first-seen order",
);
}
}