diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d822faa..915d1c8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,14 +39,12 @@ jobs: with: cache-bin: 'false' save-if: ${{ github.ref == 'refs/heads/main' }} - - name: Install mold - uses: rui314/setup-mold@339e96b4ff1f3bd865856ba4cbbe8818a7acb038 # v1 - with: - make-default: false - - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0 - with: - version: "36.2" - repo-token: ${{ secrets.GITHUB_TOKEN }} + - name: Install mold and protoc + env: + GH_TOKEN: ${{ github.token }} + run: | + sh scripts/install-ci-release-tool.sh mold + sh scripts/install-ci-release-tool.sh protoc - uses: taiki-e/install-action@1919c4a95903d2c33e4a221102edca87309fc14b # v2 with: tool: nextest @@ -78,7 +76,13 @@ jobs: with: cache-bin: 'false' save-if: ${{ github.ref == 'refs/heads/main' }} + - name: Install protoc on macOS + if: runner.os == 'macOS' + env: + GH_TOKEN: ${{ github.token }} + run: sh scripts/install-ci-release-tool.sh protoc - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3.0.0 + if: runner.os == 'Windows' with: version: "36.2" repo-token: ${{ secrets.GITHUB_TOKEN }} diff --git a/crates/omloom-contracts/build.rs b/crates/omloom-contracts/build.rs index bf705bc..b9db36e 100644 --- a/crates/omloom-contracts/build.rs +++ b/crates/omloom-contracts/build.rs @@ -18,6 +18,8 @@ fn main() -> Result<(), Box> { prost_build::Config::new() .extern_path(".omloom.v1", "crate::wire") .boxed(".omloom.workspace.v1.Network.parent") + .boxed(".omloom.workspace.v1.Query.network") + .boxed(".omloom.workspace.v1.SavedView.query") .boxed(".omloom.workspace.v1.Address.allocation") .boxed(".omloom.workspace.v1.Address.interface") .boxed(".omloom.workspace.v1.Reply.result.detail") @@ -58,7 +60,7 @@ fn main() -> Result<(), Box> { } Type::String => "Kind::Text".into(), Type::Bytes => "Kind::Bytes".into(), - Type::Uint32 => "Kind::Varint(u32::MAX as u64)".into(), + Type::Uint32 | Type::Sint32 => "Kind::Varint(u32::MAX as u64)".into(), Type::Bool => "Kind::Varint(1)".into(), Type::Enum => "Kind::Varint(i32::MAX as u64)".into(), Type::Uint64 | Type::Sint64 => "Kind::Varint(u64::MAX)".into(), @@ -81,6 +83,8 @@ fn main() -> Result<(), Box> { && matches!(field.name(), "sources" | "boundaries") { 8 + } else if message.name() == "NetworkSelection" && field.name() == "sites" { + 16 } else if (message.name() == "EvaluateRequirements" && field.name() == "nodes") || (message.name() == "RequirementGroup" && field.name() == "members") || (message.name() == "PhysicalPlan" && field.name() == "steps") @@ -94,6 +98,7 @@ fn main() -> Result<(), Box> { || (message.name() == "PhysicalCableEnd" && field.name() == "channels") || (message.name() == "PhysicalChannelSet" && field.name() == "channels") || (message.name() == "PhysicalDefinition" && field.name() == "channels") + || (message.name() == "NetworkLayout" && field.name() == "pins") { 64 } else if message.name() == "PhysicalMaterial" && field.name() == "specifications" { diff --git a/crates/omloom-contracts/src/ids.rs b/crates/omloom-contracts/src/ids.rs index 6a45d56..60f973e 100644 --- a/crates/omloom-contracts/src/ids.rs +++ b/crates/omloom-contracts/src/ids.rs @@ -67,6 +67,15 @@ pub struct AssetRef { pub asset: AssetId, pub incarnation: IncarnationId, } +impl AssetRef { + /// Fixed-width identity for database keys: asset UUID followed by incarnation UUID. + pub fn to_bytes(self) -> [u8; 32] { + let mut bytes = [0; 32]; + bytes[..16].copy_from_slice(self.asset.0.as_bytes()); + bytes[16..].copy_from_slice(self.incarnation.0.as_bytes()); + bytes + } +} impl TryFrom<&crate::wire::AssetRef> for AssetRef { type Error = Error; fn try_from(value: &crate::wire::AssetRef) -> Result { diff --git a/crates/omloom-store/src/disk.rs b/crates/omloom-store/src/disk.rs index 881b5f0..e61f8a8 100644 --- a/crates/omloom-store/src/disk.rs +++ b/crates/omloom-store/src/disk.rs @@ -190,22 +190,28 @@ fn props(items: impl IntoIterator) -> Properties { items.into_iter().map(|(k, v)| (k.into(), v)).collect() } +fn property_charge(name: &str, value: &Value, parent: &str) -> u64 { + let bytes = match value { + // Charge the maximum supported width before dispatching to a recovery owner. + Value::String(_) if parent == "leases" && name == "worker" => 256, + Value::String(value) => value.len() as u64, + Value::Binary(value) => value.len() as u64, + _ => 16, + }; + name.len() as u64 + 64 + bytes +} + fn charged_row(key: &str, parent: &str, class: Class, mut data: Properties) -> Result { + if parent == "relations" { + workspace::topology::derive_pair(&mut data)?; + } data.insert("key".into(), text(key)); data.insert("parent".into(), text(parent)); data.insert("class".into(), int(class as u64)?); let mut size = 128u64; for (name, value) in &data { - let bytes = match value { - // A recovery owner may have a longer identifier than the first - // worker. Charge its full supported width before any dispatch. - Value::String(_) if parent == "leases" && name == "worker" => 256, - Value::String(value) => value.len() as u64, - Value::Binary(value) => value.len() as u64, - _ => 16, - }; size = size - .checked_add(name.len() as u64 + 64 + bytes) + .checked_add(property_charge(name, value, parent)) .ok_or(Error::Limit)?; } if parent == "relations" { @@ -543,6 +549,7 @@ impl Store { schema.add_property(PropertyDef::new("query", PropertyType::Binary)); schema.add_property(PropertyDef::new("working", PropertyType::Binary)); schema.add_property(PropertyDef::new("payload", PropertyType::Binary)); + schema.add_property(PropertyDef::new("topology_pair", PropertyType::Binary)); for name in ["choices", "audiences", "editors", "columns"] { schema.add_property(PropertyDef::new( name, @@ -718,6 +725,7 @@ impl Store { ("omloom_asset", "asset"), // Reverse administrative inheritance follows exact relation endpoints. ("omloom_to_asset", "to_asset"), + ("omloom_topology_pair", "topology_pair"), ("omloom_group", "group"), ("omloom_party_lookup", "party_lookup"), ("omloom_request_account", "request_account"), @@ -820,6 +828,13 @@ impl Store { .db .execute_cypher_shared(query, None, None, None, None)?; } + store.db.execute_cypher_shared( + "MATCH (n:OmRow {topology_pair:$pair}) RETURN n.key AS key LIMIT 1", + params([("pair", Value::Binary(vec![0; 64]))]), + None, + None, + None, + )?; Ok(store) } fn policy_identity(&self) -> String { diff --git a/crates/omloom-store/src/disk/deltas.rs b/crates/omloom-store/src/disk/deltas.rs index 5e316ed..9c32de0 100644 --- a/crates/omloom-store/src/disk/deltas.rs +++ b/crates/omloom-store/src/disk/deltas.rs @@ -28,7 +28,15 @@ impl Capture { { return Ok(()); } + // The indexed pair is derived from the relation endpoints on every write, including undo. + let derived = old.and_then(|row| row.get("topology_pair")); let size = old.map(|p| number(p, "size")).transpose()?.unwrap_or(0); + let size = match derived { + Some(pair) => size + .checked_sub(property_charge("topology_pair", pair, "relations")) + .ok_or(Error::Corrupt)?, + None => size, + }; let charge = usize::try_from(size) .map_err(|_| Error::Limit)? .checked_mul(6) @@ -39,7 +47,13 @@ impl Capture { return Err(Error::Limit); } self.bytes = next; - self.rows.insert(key.into(), old.cloned()); + let before = old.map(|row| { + row.iter() + .filter(|(name, _)| name.as_str() != "topology_pair") + .map(|(name, value)| (name.clone(), value.clone())) + .collect() + }); + self.rows.insert(key.into(), before); Ok(()) } pub fn encode(&self) -> Result> { diff --git a/crates/omloom-store/src/disk/tests.rs b/crates/omloom-store/src/disk/tests.rs index 2846e27..7746bb1 100644 --- a/crates/omloom-store/src/disk/tests.rs +++ b/crates/omloom-store/src/disk/tests.rs @@ -147,10 +147,10 @@ fn statement_memory_exhaustion_is_a_resource_limit() { assert!(matches!(error, Error::Limit)); } -// A real old catalog differs from a new catalog with an old application marker. -// Opening must admit the additive declaration before migrating any row. +// Exercise the existing upgrade of version-2 fields in a strict catalog that +// already satisfies the initial topology schema requirement. #[test] -fn old_strict_catalog_upgrades_without_recreating_the_database() +fn old_fields_upgrade_in_a_strict_catalog_with_topology_support() -> std::result::Result<(), Box> { let dir = tempfile::tempdir()?; let policy = Policy { @@ -176,7 +176,7 @@ fn old_strict_catalog_upgrades_without_recreating_the_database() .find(|s| s.name == "OmRow") .ok_or(Error::Corrupt)?; let expected_revision = old.schema_revision; - // The complete version-2 declaration is independent of the current schema. + // Keep the version-2 fields explicit to exercise catalog admission. let mut declaration = LabelSchema::new_node_id("OmRow"); declaration.add_property(PropertyDef::new("key", PropertyType::String).not_null()); for name in [ @@ -246,6 +246,7 @@ fn old_strict_catalog_upgrades_without_recreating_the_database() )); } declaration.add_property(PropertyDef::new("digest", PropertyType::Binary)); + declaration.add_property(PropertyDef::new("topology_pair", PropertyType::Binary)); let publication = store.db.publish_schema(SchemaPackage { labels: vec![LabelDeclaration { schema: declaration, diff --git a/crates/omloom-store/src/disk/workspace.rs b/crates/omloom-store/src/disk/workspace.rs index 143d4d9..d3d6476 100644 --- a/crates/omloom-store/src/disk/workspace.rs +++ b/crates/omloom-store/src/disk/workspace.rs @@ -24,6 +24,7 @@ mod responsibility; mod segments; mod software; mod source_layout; +pub(super) mod topology; mod typed_values; mod views; mod vm_placement; @@ -444,7 +445,15 @@ impl Store { fn workspace_rows(&self, parent: &str, limit: usize) -> Result> { let tx = self.begin()?; - let rows = tx.query( + let rows = tx.workspace_rows(parent, limit)?; + tx.commit()?; + Ok(rows) + } +} + +impl Tx<'_> { + fn workspace_rows(&self, parent: &str, limit: usize) -> Result> { + let rows = self.query( "MATCH (n:OmRow) WHERE n.parent=$parent RETURN properties(n) AS p LIMIT $limit", params([ ("parent", text(parent)), @@ -454,12 +463,8 @@ impl Store { if rows.len() > limit { return Err(Error::Limit); } - tx.commit()?; Ok(rows) } -} - -impl Tx<'_> { fn raw( &self, query: &str, diff --git a/crates/omloom-store/src/disk/workspace/assets.rs b/crates/omloom-store/src/disk/workspace/assets.rs index 6f97e2b..43021cd 100644 --- a/crates/omloom-store/src/disk/workspace/assets.rs +++ b/crates/omloom-store/src/disk/workspace/assets.rs @@ -115,7 +115,7 @@ pub(super) fn ownership_path(children: bool, reverse: bool) -> &'static str { } } -fn browse_clause( +pub(super) fn browse_clause( context: &Context, values: &mut HashMap, reverse: Option, @@ -512,6 +512,16 @@ impl Store { return Err(Error::Invalid); } let view = api::View::try_from(query.view).map_err(|_| Error::Invalid)?; + let include_topology = query + .network + .as_ref() + .is_some_and(|network| network.include_topology); + if include_topology && view != api::View::Network { + return Err(Error::Invalid); + } + if let Some(network) = &mut query.network { + network.layout = None; + } if let Some(filter) = &query.responsibility { super::responsibility::validate_filter(filter)?; } @@ -542,6 +552,31 @@ impl Store { "a.parent='assets'".to_owned(), visible_clause(context, None, &mut values), ]; + let mut site_entry = None; + let without_site = query + .network + .as_ref() + .is_some_and(|network| network.unplaced); + if let Some(network) = &query.network + && !network.sites.is_empty() + { + if without_site { + return Err(Error::Invalid); + } + tx.validate_sites(context, &network.sites)?; + let members = tx.working_scope(context, &network.sites)?; + let sites = parameter( + &mut values, + Value::Array( + members + .into_iter() + .map(|target| text(asset_key(target))) + .collect(), + ), + ); + clauses.push(format!("a.key IN {sites}")); + site_entry = Some(sites); + } if query.require_physical_read { clauses.push(visible_clause(context, Some("physical"), &mut values)); } @@ -660,13 +695,20 @@ impl Store { focus: query.focus.clone(), expected: None, }; - let browse_roots: Vec<_> = query + if view == api::View::Network && !query.level_only && working.focus.is_some() { + if !working.roots.is_empty() { + // Validate the branch against the project, including IPAM containment. + tx.selection(context, &working, false)?; + } + working.roots = working.focus.take().into_iter().collect(); + } + let browse_roots: Vec<_> = working .focus .iter() - .chain(query.roots.iter().filter(|_| query.focus.is_none())) + .chain(working.roots.iter().filter(|_| working.focus.is_none())) .cloned() .collect(); - let mut level_only = query.level_only || query.focus.is_some(); + let mut level_only = query.level_only || working.focus.is_some(); for root in &browse_roots { let row = tx.visible_asset(context, parse_target(Some(root))?)?; if [ @@ -1204,6 +1246,11 @@ impl Store { clauses.push(format!("a.key IN {keys}")); } } + if let Some(keys) = site_entry + && entry == "MATCH (a:OmRow {parent:'assets'})" + { + entry = format!("UNWIND {keys} AS site_key MATCH (a:OmRow {{key:site_key}})"); + } clauses.retain(|clause| clause != "true"); let tail = format!( "{} WITH {} WHERE {}", @@ -1211,13 +1258,23 @@ impl Store { bindings.join(", "), clauses.join(" AND ") ); - let bases = match prefix_entries { + let mut bases = match prefix_entries { Some(entries) => entries .into_iter() .map(|entry| format!("{entry}{tail}")) .collect::>(), None => vec![format!("{entry}{tail}")], }; + if without_site { + let keys = tx.unplaced_candidates(context, &bases, &values)?; + let keys = parameter( + &mut values, + Value::Array(keys.into_iter().map(text).collect()), + ); + for base in &mut bases { + base.push_str(&format!(" AND a.key IN {keys}")); + } + } let sort = match query.sort.as_str() { "" | "name" => format!( "CASE WHEN {} THEN a.name_sort ELSE a.asset END", @@ -1399,7 +1456,8 @@ impl Store { // Reserve both cohort positions when a page merges disjoint sources. let cursor_bytes = 2 * 20 + 3 + fingerprint.len() + selected.position.max_bytes(); bytes += 1 + prost::length_delimiter_len(cursor_bytes) + cursor_bytes; - let envelope = 1 + prost::length_delimiter_len(MAX_REPLY_BYTES); + let envelope = + 1 + prost::length_delimiter_len(MAX_REPLY_BYTES) + if include_topology { 8 } else { 0 }; let mut emitted = 0; for (index, item) in items.iter().enumerate() { let size = item.encoded_len(); @@ -1421,6 +1479,11 @@ impl Store { if query.include_context { contexts.truncate(emitted); } + let topology = if include_topology { + Some(tx.network_topology(context, &items, MAX_REPLY_BYTES - envelope - bytes)?) + } else { + None + }; tx.commit()?; if self.usage()?.revision != revision { return Err(Error::Expired); @@ -1429,6 +1492,7 @@ impl Store { Ok(api::Page { items, contexts, + topology, total, revision, next: if next_position.rank()? < total { diff --git a/crates/omloom-store/src/disk/workspace/exchange.rs b/crates/omloom-store/src/disk/workspace/exchange.rs index dc43cbb..6bcb859 100644 --- a/crates/omloom-store/src/disk/workspace/exchange.rs +++ b/crates/omloom-store/src/disk/workspace/exchange.rs @@ -455,6 +455,9 @@ impl Store { ) -> Result { let mut query = export.query.ok_or(Error::Invalid)?; query.cursor = None; + if let Some(network) = &mut query.network { + network.include_topology = false; + } query.limit = MAX_PAGE as u32; let mut columns = query.columns.clone(); if columns.is_empty() { diff --git a/crates/omloom-store/src/disk/workspace/segments/projection.rs b/crates/omloom-store/src/disk/workspace/segments/projection.rs index f27d334..0487961 100644 --- a/crates/omloom-store/src/disk/workspace/segments/projection.rs +++ b/crates/omloom-store/src/disk/workspace/segments/projection.rs @@ -247,6 +247,7 @@ impl Store { )?; tx.commit()?; Ok(reply(api::reply::Result::Page(api::Page { + topology: None, items, total: total as u64, revision, diff --git a/crates/omloom-store/src/disk/workspace/topology.rs b/crates/omloom-store/src/disk/workspace/topology.rs new file mode 100644 index 0000000..2af5f60 --- /dev/null +++ b/crates/omloom-store/src/disk/workspace/topology.rs @@ -0,0 +1,304 @@ +use super::*; +use wire::RelationKind as R; + +const MAX_EDGES: usize = 128; +const RELATIONS: &[R] = &[ + R::MemberOf, + R::On, + R::AssignedTo, + R::Contains, + R::Carries, + R::ConnectedTo, + R::Uses, +]; + +fn pair_key(from: AssetRef, to: AssetRef) -> Value { + let mut bytes = Vec::with_capacity(64); + bytes.extend_from_slice(&from.to_bytes()); + bytes.extend_from_slice(&to.to_bytes()); + Value::Binary(bytes) +} + +pub(in crate::disk) fn derive_pair(data: &mut Properties) -> Result<()> { + let kind = R::try_from(number(data, "kind")? as i32).map_err(|_| Error::Corrupt)?; + if RELATIONS.contains(&kind) { + let from = target(data)?; + let to = AssetRef { + asset: string(data, "to_asset")? + .parse() + .map_err(|_| Error::Corrupt)?, + incarnation: string(data, "to_incarnation")? + .parse() + .map_err(|_| Error::Corrupt)?, + }; + data.insert("topology_pair".into(), pair_key(from, to)); + } else { + data.remove("topology_pair"); + } + Ok(()) +} + +impl Tx<'_> { + /// Project configured participants by exact membership keys, without expanding other interfaces. + fn topology_assignments( + &self, + context: &Context, + items: &[api::Item], + ) -> Result> { + use wire::AssetKind as K; + let mut owners = Vec::new(); + let mut interfaces = Vec::new(); + let mut allocation_keys = Vec::new(); + for item in items { + let target = parse_target(item.target.as_ref())?; + if !context.field(target, item.kind, "relations", false) { + continue; + } + match K::try_from(item.kind) { + Ok(K::Interface) => interfaces.push(target), + Ok(K::IpAddress | K::AddressGroup) + if context.field(target, item.kind, "network", false) => + { + if item.kind == K::IpAddress as i32 { + allocation_keys + .push(text(format!("network/{}", network::identity(target)))); + } else { + owners.push((target, target)); + } + } + _ => {} + } + } + if interfaces.is_empty() || (owners.is_empty() && allocation_keys.is_empty()) { + return Ok(Vec::new()); + } + if !allocation_keys.is_empty() { + let rows = self.query( + "UNWIND $keys AS key MATCH (n:OmRow {key:key}) RETURN properties(n) AS p", + params([("keys", Value::Array(allocation_keys))]), + )?; + for row in rows { + let address = target(&row)?; + let owner = network::groups::group_of(&row)?.unwrap_or(address); + if owner != address { + let kind = K::AddressGroup as i32; + if !context.field(owner, kind, "network", false) + || number(&self.asset(owner)?, "adoption")? + != wire::Adoption::Adopted as u64 + { + continue; + } + } + owners.push((address, owner)); + } + } + let pairs: Vec<_> = owners + .iter() + .flat_map(|&(address, owner)| { + let parent = format!("members/{}", network::identity(owner)); + interfaces.iter().map(move |&interface| { + ( + address, + interface, + format!("{parent}/{}", network::identity(interface)), + ) + }) + }) + .collect(); + let keys: BTreeSet<_> = pairs.iter().map(|(_, _, key)| key.as_str()).collect(); + if keys.is_empty() { + return Ok(Vec::new()); + } + let rows = self.raw( + "UNWIND $keys AS key MATCH (n:OmRow {key:key}) RETURN n.key AS key", + [( + "keys".into(), + Value::Array(keys.into_iter().map(text).collect()), + )] + .into(), + )?; + let existing = rows + .iter() + .map(|row| row.get("key").and_then(Value::as_str).ok_or(Error::Corrupt)) + .collect::>>()?; + Ok(pairs + .into_iter() + .filter(|(_, _, key)| existing.contains(key.as_str())) + .take(MAX_EDGES + 1) + .map(|(from, to, _)| api::NetworkLink { + relation_id: format!( + "assignment/{}/{}", + network::identity(from), + network::identity(to) + ), + from: Some(target_wire(from)), + to: Some(target_wire(to)), + kind: wire::RelationKind::AssignedTo as i32, + }) + .collect()) + } + + pub(super) fn validate_sites(&self, context: &Context, sites: &[wire::AssetRef]) -> Result<()> { + if sites.len() > 16 { + return Err(Error::Limit); + } + for site in sites { + let target = parse_target(Some(site))?; + let row = self.visible_asset(context, target)?; + if number(&row, "kind")? != wire::AssetKind::Site as u64 { + return Err(Error::Invalid); + } + } + Ok(()) + } + + /// Search upward only from candidates admitted by the other view filters. + pub(super) fn unplaced_candidates( + &self, + context: &Context, + bases: &[String], + values: &HashMap, + ) -> Result> { + const CANDIDATES: usize = 4096; + const ANCESTORS: usize = 8192; + const BATCH: usize = 128; + const BATCH_EDGES: usize = 2048; + let site = wire::AssetKind::Site as i32; + let mut candidates = BTreeSet::new(); + let mut located = BTreeSet::new(); + for base in bases { + let rows = self.raw( + &format!("{base} RETURN DISTINCT a.key AS key, a.asset AS asset, a.incarnation AS incarnation, a.kind AS kind LIMIT {}", CANDIDATES + 1), + values.clone(), + )?; + for row in rows { + let key = string(&row, "key")?; + if number(&row, "kind")? == site as u64 + && context.field(target(&row)?, site, "relations", false) + { + located.insert(key.clone()); + } + candidates.insert(key); + if candidates.len() > CANDIDATES { + return Err(Error::Limit); + } + } + } + let mut visited = candidates.clone(); + let mut queue: Vec<_> = candidates.difference(&located).cloned().collect(); + let mut children = BTreeMap::>::new(); + let mut work = 0; + while !queue.is_empty() { + let start = queue.len() - queue.len().min(BATCH); + let keys = queue.split_off(start); + for reverse in [false, true] { + let mut values = HashMap::new(); + let owned = assets::browse_clause(context, &mut values, Some(reverse), &[]); + let path = assets::ownership_path(false, reverse); + values.insert("keys".into(), Value::Array(keys.iter().map(text).collect())); + let rows = self.raw(&format!( + "UNWIND $keys AS key MATCH {path} WHERE {owned} RETURN DISTINCT key AS child,p.key AS parent,p.kind AS kind LIMIT {}", BATCH_EDGES + 1 + ), values)?; + if rows.len() > BATCH_EDGES { + return Err(Error::Limit); + } + work += rows.len(); + if work > ANCESTORS { + return Err(Error::Limit); + } + for row in rows { + let parent = string(&row, "parent")?; + children + .entry(parent.clone()) + .or_default() + .insert(string(&row, "child")?); + if number(&row, "kind")? == site as u64 { + located.insert(parent); + } else if visited.insert(parent.clone()) { + queue.push(parent); + } + } + } + } + let mut queue: Vec<_> = located.iter().cloned().collect(); + while let Some(parent) = queue.pop() { + if let Some(descendants) = children.remove(&parent) { + for child in descendants { + if located.insert(child.clone()) { + queue.push(child); + } + } + } + } + candidates.retain(|key| !located.contains(key)); + Ok(candidates) + } + + pub(super) fn network_topology( + &self, + context: &Context, + items: &[api::Item], + available_bytes: usize, + ) -> Result { + let mut authorized = Vec::with_capacity(items.len()); + for item in items { + let target = parse_target(item.target.as_ref())?; + if context.field(target, item.kind, "relations", false) { + authorized.push(( + target, + wire::AssetKind::try_from(item.kind).map_err(|_| Error::Corrupt)?, + )); + } + } + let mut topology = api::NetworkTopology::default(); + // The initial schema includes the pair index, and relation writes supply its keys. + // Exact pair probes keep off-page adjacency out of the read even for hubs. + let mut pairs = Vec::new(); + for &(from, source_kind) in &authorized { + for &(to, target_kind) in &authorized { + if RELATIONS + .iter() + .any(|&kind| inventory::compatible(source_kind, target_kind, kind)) + { + pairs.push(pair_key(from, to)); + } + } + } + let rows = if pairs.is_empty() { + Vec::new() + } else { + self.query( + "UNWIND $pairs AS pair MATCH (r:OmRow {topology_pair:pair}) RETURN properties(r) AS p ORDER BY r.key LIMIT $limit", + params([ + ("pairs", Value::Array(pairs)), + ("limit", Value::Int((MAX_EDGES + 1) as i64)), + ]), + )? + }; + let mut edges = Vec::with_capacity(rows.len()); + for row in rows { + let relation = super::super::inventory::relation_projection(&row)?; + edges.push(api::NetworkLink { + relation_id: relation.relation_id, + from: relation.from, + to: relation.to, + kind: relation.kind, + }); + } + if edges.len() <= MAX_EDGES { + edges.extend(self.topology_assignments(context, items)?); + } + let mut remaining = available_bytes; + for edge in edges { + let size = edge.encoded_len(); + let bytes = 1 + prost::length_delimiter_len(size) + size; + if topology.edges.len() == MAX_EDGES || bytes > remaining { + topology.more_edges = true; + break; + } + remaining -= bytes; + topology.edges.push(edge); + } + Ok(topology) + } +} diff --git a/crates/omloom-store/src/disk/workspace/views.rs b/crates/omloom-store/src/disk/workspace/views.rs index 96bc304..809d5da 100644 --- a/crates/omloom-store/src/disk/workspace/views.rs +++ b/crates/omloom-store/src/disk/workspace/views.rs @@ -15,21 +15,106 @@ fn decode(row: &Properties) -> Result { name: string(row, "label")?, owner: string(row, "owner")?, revision: number(row, "revision")?, - query: Some(api::Query::decode(binary(row, "query")?).map_err(|_| Error::Corrupt)?), + query: Some(Box::new( + api::Query::decode(binary(row, "query")?).map_err(|_| Error::Corrupt)?, + )), audiences: read_strings(row, "audiences")?, editors: read_strings(row, "editors")?, }) } +fn visible_layouts<'a>( + tx: &Tx<'_>, + context: &Context, + queries: impl IntoIterator, +) -> Result> { + let mut keys = BTreeSet::new(); + let mut layouts = Vec::new(); + for query in queries { + let Some(layout) = query + .network + .as_mut() + .and_then(|network| network.layout.as_mut()) + else { + continue; + }; + if query.view != api::View::Network as i32 || layout.pins.len() > 64 { + return Err(Error::Limit); + } + let mut targets = Vec::with_capacity(layout.pins.len()); + for pin in &layout.pins { + let target = parse_target(pin.target.as_ref())?; + if targets.contains(&target) + || !(-4096..=4096).contains(&pin.x) + || !(-4096..=4096).contains(&pin.y) + { + return Err(Error::Invalid); + } + keys.insert(asset_key(target)); + targets.push(target); + } + layouts.push((layout, targets)); + } + let mut permitted = BTreeSet::new(); + let mut current = BTreeSet::new(); + let mut keys = keys.into_iter(); + loop { + let batch: Vec<_> = keys.by_ref().take(128).map(text).collect(); + if batch.is_empty() { + break; + } + let rows = tx.query( + "UNWIND $keys AS key MATCH (a:OmRow {key:key}) WHERE a.adoption=$adopted RETURN {asset:a.asset,incarnation:a.incarnation,kind:a.kind} AS p LIMIT 128", + params([ + ("keys", Value::Array(batch)), + ("adopted", Value::Int(wire::Adoption::Adopted as i64)), + ]), + )?; + for row in rows { + let reference = target(&row)?; + current.insert(reference); + if context.readable(reference, number(&row, "kind")? as i32) { + permitted.insert(reference); + } + } + } + for (layout, targets) in layouts { + let mut targets = targets.into_iter(); + layout.pins.retain(|_| { + targets + .next() + .is_some_and(|target| permitted.contains(&target)) + }); + } + Ok(current) +} + +fn pins(query: &api::Query) -> &[api::NetworkPin] { + query + .network + .as_ref() + .and_then(|network| network.layout.as_ref()) + .map_or(&[], |layout| layout.pins.as_slice()) +} + impl Store { pub(super) fn workspace_views(&self, context: &Context) -> Result> { - self.workspace_rows("views", 256)? - .iter() - .map(decode) - .filter(|r| match r { - Err(_) => true, - Ok(v) => visible(v, context), - }) - .collect() + let tx = self.begin()?; + let mut views = Vec::new(); + for row in tx.workspace_rows("views", 256)? { + let view = decode(&row)?; + if visible(&view, context) { + views.push(view); + } + } + visible_layouts( + &tx, + context, + views + .iter_mut() + .filter_map(|view| view.query.as_deref_mut()), + )?; + tx.commit()?; + Ok(views) } pub(super) fn workspace_save_view( &self, @@ -38,6 +123,21 @@ impl Store { now: u64, ) -> Result { let mut view = request.view.ok_or(Error::Invalid)?; + if request.remove { + let id = view.id.parse::().map_err(|_| Error::Invalid)?; + let mut tx = self.begin()?; + let key = format!("view/{id}"); + let row = tx.get(&key)?.ok_or(Error::Missing)?; + if string(&row, "owner")? != context.actor { + return Err(Error::Denied); + } + if number(&row, "revision")? != view.revision { + return Err(Error::Conflict); + } + tx.delete(&key)?; + tx.audit(&key, &context.actor, "view.removed", view.revision, now)?; + return Ok(mutation(tx.commit()?, Vec::new())); + } if view.name.trim().is_empty() || view.name.len() > 100 || view.audiences.len() > 16 @@ -47,6 +147,12 @@ impl Store { } let query = view.query.as_mut().ok_or(Error::Invalid)?; query.cursor = None; + if let Some(network) = &mut query.network { + network.include_topology = false; + if network.sites.is_empty() && !network.unplaced && network.layout.is_none() { + query.network = None; + } + } if query.encoded_len() > 8192 || query.predicates.len() > 8 || query.columns.len() > MAX_FIELDS @@ -66,6 +172,14 @@ impl Store { super::responsibility::validate_filter(filter)?; } let mut tx = self.begin()?; + if let Some(network) = &query.network + && !network.sites.is_empty() + { + if network.unplaced { + return Err(Error::Invalid); + } + tx.validate_sites(context, &network.sites)?; + } let old = if view.id.is_empty() { if view.revision != 0 || (!view.owner.is_empty() && view.owner != context.actor) { return Err(Error::Denied); @@ -80,9 +194,9 @@ impl Store { .ok_or(Error::Missing)?, ) }; - if let Some(row) = &old { - let previous = decode(row)?; - if !visible(&previous, context) || !editable(&previous, context) { + let previous = old.as_ref().map(decode).transpose()?; + if let Some(previous) = &previous { + if !visible(previous, context) || !editable(previous, context) { return Err(Error::Denied); } if previous.revision != view.revision { @@ -97,23 +211,23 @@ impl Store { if previous.owner != view.owner && !context.recipients.contains(&view.owner) { return Err(Error::Denied); } - if request.remove { - if previous.owner != context.actor { - return Err(Error::Denied); + } + let previous_query = previous.as_ref().and_then(|saved| saved.query.as_deref()); + let mut readable_previous = previous_query.filter(|q| !pins(q).is_empty()).cloned(); + let current = visible_layouts( + &tx, + context, + core::iter::once(query.as_mut()).chain(readable_previous.iter_mut()), + )?; + let mut hidden = Vec::new(); + if let (Some(previous), Some(readable)) = (previous_query, readable_previous.as_ref()) { + for pin in pins(previous) { + if current.contains(&parse_target(pin.target.as_ref())?) + && !pins(readable).contains(pin) + { + hidden.push(pin.clone()); } - let key = format!("view/{}", view.id); - tx.delete(&key)?; - tx.audit(&key, &context.actor, "view.removed", view.revision, now)?; - return Ok(mutation(tx.commit()?, Vec::new())); - } - if previous == view { - tx.commit()?; - return Ok(reply(api::reply::Result::Views(api::Views { - items: vec![view], - }))); - } - } else if request.remove { - return Err(Error::Invalid); + } } let existing_audiences = old .as_ref() @@ -132,9 +246,25 @@ impl Store { || old .as_ref() .is_some_and(|p| string(p, "owner").ok().as_ref() != Some(&view.owner)); + if !hidden.is_empty() + && (view + .audiences + .iter() + .any(|a| !existing_audiences.contains(a)) + || previous + .as_ref() + .is_some_and(|saved| saved.owner != view.owner)) + { + return Err(Error::Denied); + } if disclosure && request.publish { let query = view.query.as_ref().ok_or(Error::Invalid)?; - for reference in query.roots.iter().chain(query.focus.iter()) { + for reference in query + .roots + .iter() + .chain(query.focus.iter()) + .chain(pins(query).iter().filter_map(|pin| pin.target.as_ref())) + { let target = parse_target(Some(reference))?; let row = tx.visible_asset(context, target)?; if number(&row, "kind")? == wire::AssetKind::ProviderAccount as u64 @@ -155,8 +285,45 @@ impl Store { items: vec![view], }))); } + let mut stored = view.clone(); + if !hidden.is_empty() { + let query = stored.query.as_mut().ok_or(Error::Invalid)?; + if query.view != api::View::Network as i32 { + return Err(Error::Denied); + } + let layout = query + .network + .get_or_insert_with(Default::default) + .layout + .get_or_insert_with(Default::default); + if layout.pins.len() + hidden.len() > 64 { + return Err(Error::Limit); + } + if let (Some(previous), Some(readable)) = (previous_query, readable_previous.as_ref()) + && layout.pins == pins(readable) + { + // Unchanged visible positions keep their original interleaving with hidden pins. + layout.pins = pins(previous) + .iter() + .filter(|pin| hidden.contains(pin) || pins(readable).contains(pin)) + .cloned() + .collect(); + } else { + layout.pins.extend(hidden); + } + if query.encoded_len() > 8192 { + return Err(Error::Limit); + } + } + if previous.as_ref() == Some(&stored) { + tx.commit()?; + return Ok(reply(api::reply::Result::Views(api::Views { + items: vec![view], + }))); + } view.revision = view.revision.checked_add(1).ok_or(Error::Limit)?; - tx.require_view_capacity(&view)?; + stored.revision = view.revision; + tx.require_view_capacity(&stored)?; let key = format!("view/{}", view.id); tx.put_against( &key, @@ -170,7 +337,7 @@ impl Store { ("editors", strings(&view.editors)), ( "query", - Value::Binary(view.query.as_ref().ok_or(Error::Invalid)?.encode_to_vec()), + Value::Binary(stored.query.as_ref().ok_or(Error::Invalid)?.encode_to_vec()), ), ]), old.as_ref(), diff --git a/crates/omloom-store/src/disk/workspace/working.rs b/crates/omloom-store/src/disk/workspace/working.rs index 2b1169d..26ed119 100644 --- a/crates/omloom-store/src/disk/workspace/working.rs +++ b/crates/omloom-store/src/disk/workspace/working.rs @@ -418,24 +418,36 @@ impl Tx<'_> { return Err(Error::Limit); } let mut selected = BTreeSet::new(); - let mut pending = Vec::new(); + let mut frontier = BTreeSet::new(); for root in roots { let target = parse_target(Some(root))?; self.visible_asset(context, target)?; if selected.insert(target) { - pending.push(target); + frontier.insert(asset_key(target)); } } - while let Some(parent) = pending.pop() { - for child in self.local_children(context, parent)? { - if !selected.contains(&child) { + while !frontier.is_empty() { + let mut next = BTreeSet::new(); + for (parent, child, relation) in self.context_edges(context, &frontier, true)? { + let child_ref = crate::disk::target(&child)?; + let parent_ref = crate::disk::target(&parent)?; + let parent_kind = wire::AssetKind::try_from(number(&parent, "kind")? as i32) + .map_err(|_| Error::Corrupt)?; + let child_kind = wire::AssetKind::try_from(number(&child, "kind")? as i32) + .map_err(|_| Error::Corrupt)?; + let from = parse_target(relation.from.as_ref())?; + let to = parse_target(relation.to.as_ref())?; + if refines(parent_ref, parent_kind, child_kind, relation.kind, from, to) + && !selected.contains(&child_ref) + { if selected.len() >= 1024 { return Err(Error::Limit); } - selected.insert(child); - pending.push(child); + selected.insert(child_ref); + next.insert(asset_key(child_ref)); } } + frontier = next; } Ok(selected) } diff --git a/crates/omloom-store/tests/workspace.rs b/crates/omloom-store/tests/workspace.rs index 588f10d..7fdfe64 100644 --- a/crates/omloom-store/tests/workspace.rs +++ b/crates/omloom-store/tests/workspace.rs @@ -42,6 +42,8 @@ mod responsibility; mod segments; #[path = "workspace/software.rs"] mod software; +#[path = "workspace/topology.rs"] +mod topology; #[path = "workspace/typed_values.rs"] mod typed_values; #[path = "workspace/views.rs"] diff --git a/crates/omloom-store/tests/workspace/accounts.rs b/crates/omloom-store/tests/workspace/accounts.rs index 5b5e16b..7519ecc 100644 --- a/crates/omloom-store/tests/workspace/accounts.rs +++ b/crates/omloom-store/tests/workspace/accounts.rs @@ -896,13 +896,13 @@ fn account_card_sharing_requires_its_own_capability() -> TestResult { let (_, account) = setup(&store)?; let mut actor = operator(); actor.grants[0].write = Fields::Selected(vec!["account.use".into()]); - let publish = api::SaveView { + let mut publish = api::SaveView { view: Some(api::SavedView { name: "Selected account".into(), - query: Some(api::Query { - focus: Some(account), + query: Some(Box::new(api::Query { + focus: Some(account.clone()), ..query() - }), + })), audiences: vec!["operations".into()], ..Default::default() }), @@ -915,6 +915,31 @@ fn account_card_sharing_requires_its_own_capability() -> TestResult { Err(Error::Denied) )); assert_eq!(store.usage()?, before); + let query = publish + .view + .as_mut() + .expect("view") + .query + .as_mut() + .expect("query"); + query.focus = None; + query.view = api::View::Network as i32; + query.network = Some(Box::new(api::NetworkSelection { + layout: Some(api::NetworkLayout { + pins: vec![api::NetworkPin { + target: Some(account), + x: 10, + y: 10, + }], + ..Default::default() + }), + ..Default::default() + })); + assert!(matches!( + call(&store, &actor, Command::SaveView(publish.clone())), + Err(Error::Denied) + )); + assert_eq!(store.usage()?, before); actor.grants[0].write = Fields::Selected(vec!["account.share".into()]); let Reply::Views(published) = call(&store, &actor, Command::SaveView(publish))? else { panic!("view"); diff --git a/crates/omloom-store/tests/workspace/topology.rs b/crates/omloom-store/tests/workspace/topology.rs new file mode 100644 index 0000000..358e4d5 --- /dev/null +++ b/crates/omloom-store/tests/workspace/topology.rs @@ -0,0 +1,691 @@ +use super::*; +use api::{reply::Result as R, request::Command as C}; + +fn graph(store: &Store, actor: &Context, query: api::Query) -> Result { + match call(store, actor, C::Query(query))? { + R::Page(page) => Ok(page), + _ => Err(Error::Corrupt), + } +} + +fn graph_query() -> api::Query { + api::Query { + view: api::View::Network as i32, + network: Some(Box::new(api::NetworkSelection { + include_topology: true, + ..Default::default() + })), + ..query() + } +} + +// Links and their continuation share a current cut with the permitted node page. +#[test] +fn topology_projects_known_links_without_disclosing_denied_endpoints() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let Some(R::Mutation(_)) = store + .workspace( + &actor, + api::Request { + command: Some(C::Address(api::Address { + address: Some("192.0.2.9/24".into()), + create_prefix: true, + ..Default::default() + })), + working: Some(api::WorkingContext { + expected: Some(store.usage()?.revision), + ..Default::default() + }), + }, + 100, + )? + .result + else { + return Err("known address result".into()); + }; + let before = store.usage()?; + let page = graph(&store, &actor, graph_query())?; + assert_eq!(page.total, 2); + let topology = page.topology.ok_or("topology")?; + assert_eq!(topology.edges.len(), 1); + assert_eq!(topology.edges[0].kind, wire::RelationKind::MemberOf as i32); + for edge in &topology.edges { + assert!(page.items.iter().any(|item| item.target == edge.from)); + assert!(page.items.iter().any(|item| item.target == edge.to)); + } + assert!(!topology.more_edges); + let mut reader = operator(); + reader.grants[0].read = Fields::Selected(vec!["name".into()]); + let restricted = graph(&store, &reader, graph_query())?; + let topology = restricted.topology.ok_or("restricted topology")?; + assert!(topology.edges.is_empty()); + assert!(!topology.more_edges); + reader.grants[0].scope = Scope::Kinds(vec![K::Prefix]); + let restricted = graph(&store, &reader, graph_query())?; + assert_eq!(restricted.total, 1); + assert!( + restricted + .topology + .ok_or("restricted topology")? + .edges + .is_empty() + ); + assert_eq!(store.usage()?, before); + + let mut selector = graph_query(); + selector.limit = 1; + let first = graph(&store, &actor, selector.clone())?; + assert_eq!(first.items.len(), 1); + assert!(first.topology.ok_or("first topology")?.edges.is_empty()); + selector.cursor = first.next; + assert!(selector.cursor.is_some()); + let second = graph(&store, &actor, selector.clone())?; + assert_eq!(second.items.len(), 1); + assert_ne!(second.items[0].target, first.items[0].target); + assert!(second.topology.ok_or("second topology")?.edges.is_empty()); + create(&store, &actor, K::Vrf, "New revision")?; + assert!(matches!( + graph(&store, &actor, selector), + Err(Error::Expired) + )); + Ok(()) +} + +fn child( + store: &Store, + actor: &Context, + parent: &wire::AssetRef, + kind: K, + name: &str, +) -> Result { + let response = store.workspace( + actor, + api::Request { + command: Some(C::Write(api::Write { + kind: kind as i32, + parent: Some(parent.clone()), + changes: vec![text_field("name", name)], + ..Default::default() + })), + working: Some(api::WorkingContext { + focus: Some(parent.clone()), + expected: Some(store.usage()?.revision), + ..Default::default() + }), + }, + 100, + )?; + let Some(R::Mutation(mutation)) = response.result else { + return Err(Error::Corrupt); + }; + mutation.targets.into_iter().next().ok_or(Error::Corrupt) +} + +// Location intersects project selection before count and page; absence stays explicit. +#[test] +fn topology_intersects_sites_with_independent_selection() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let east = create(&store, &actor, K::Site, "East")?; + let west = create(&store, &actor, K::Site, "West")?; + let east_host = child(&store, &actor, &east, K::Device, "East host")?; + let west_host = child(&store, &actor, &west, K::Device, "West host")?; + let east_port = child(&store, &actor, &east_host, K::Interface, "East interface")?; + let west_port = child(&store, &actor, &west_host, K::Interface, "West interface")?; + let unplaced_host = create(&store, &actor, K::Device, "Unplaced host")?; + let unplaced = child( + &store, + &actor, + &unplaced_host, + K::Interface, + "Unplaced interface", + )?; + let mut selector = graph_query(); + selector.kinds = vec![K::Interface as i32]; + assert_eq!(graph(&store, &actor, selector.clone())?.total, 3); + selector.roots = vec![east_host, west_host]; + assert_eq!(graph(&store, &actor, selector.clone())?.total, 2); + selector.network.as_mut().ok_or("selection")?.sites = vec![east.clone()]; + let page = graph(&store, &actor, selector.clone())?; + assert_eq!(page.total, 1); + assert_eq!(page.items[0].target.as_ref(), Some(&east_port)); + selector.roots = vec![west_port, unplaced]; + assert_eq!(graph(&store, &actor, selector.clone())?.total, 0); + let mut denied = operator(); + denied.grants[0].scope = Scope::Kinds(vec![K::Interface]); + assert!(matches!( + graph(&store, &denied, selector), + Err(Error::Denied) + )); + let mut unknown = graph_query(); + unknown.kinds = vec![K::Interface as i32]; + unknown.network.as_mut().ok_or("network")?.unplaced = true; + assert_eq!(graph(&store, &actor, unknown)?.total, 1); + Ok(()) +} + +// Selecting a network branch includes its deeper permitted interfaces and keeps the enclosing project. +#[test] +fn network_branch_queries_nested_resources_within_enclosing_roots() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let site = create(&store, &actor, K::Site, "Production site")?; + let host = child(&store, &actor, &site, K::Device, "Production switch")?; + let interface = child(&store, &actor, &host, K::Interface, "Production uplink")?; + let other = create(&store, &actor, K::Site, "Other site")?; + let outside = child(&store, &actor, &other, K::Device, "Other switch")?; + child(&store, &actor, &outside, K::Interface, "Other uplink")?; + let mut selector = graph_query(); + selector.roots = vec![site.clone()]; + selector.focus = Some(site); + selector.kinds = vec![K::Interface as i32]; + selector.level_only = false; + let page = graph(&store, &actor, selector.clone())?; + assert_eq!(page.total, 1); + assert_eq!(page.items[0].target.as_ref(), Some(&interface)); + selector.level_only = true; + assert_eq!(graph(&store, &actor, selector.clone())?.total, 0); + selector.level_only = false; + selector.focus = Some(other); + assert!(matches!( + graph(&store, &actor, selector), + Err(Error::Denied) + )); + Ok(()) +} + +// A project rooted at an address space allows navigation by namespace and range. +#[test] +fn network_branch_accepts_prefix_within_project_address_space() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let space = create(&store, &actor, K::AddressSpace, "Project space")?; + let other = create(&store, &actor, K::AddressSpace, "Other space")?; + let network = |kind: K, cidr: &str, space: &wire::AssetRef| match call( + &store, + &actor, + C::Network(api::Network { + kind: kind as i32, + name: cidr.into(), + cidr: cidr.into(), + address_space: Some(space.clone()), + sharing: api::SharedUse::Exclusive as i32, + ..Default::default() + }), + )? { + R::Mutation(result) => result.targets.into_iter().next().ok_or(Error::Corrupt), + _ => Err(Error::Corrupt), + }; + let prefix = network(K::Prefix, "192.0.2.0/25", &space)?; + let outside = network(K::Prefix, "192.0.2.0/25", &other)?; + let address = network(K::IpAddress, "192.0.2.3/25", &space)?; + network(K::IpAddress, "192.0.2.4/25", &other)?; + network(K::IpAddress, "192.0.2.130/25", &space)?; + + let mut selector = graph_query(); + selector.roots = vec![space]; + selector.focus = Some(prefix); + selector.kinds = vec![K::IpAddress as i32]; + selector.level_only = false; + let before = store.usage()?; + let page = graph(&store, &actor, selector.clone())?; + assert_eq!(page.total, 1); + assert_eq!(page.items[0].target.as_ref(), Some(&address)); + selector.focus = Some(outside); + assert!(matches!( + graph(&store, &actor, selector), + Err(Error::Denied) + )); + assert_eq!(store.usage()?, before); + Ok(()) +} + +// Filtering for one interface must not consume the unrelated site inventory. +#[test] +fn topology_unplaced_narrow_query_ignores_unrelated_sites() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let site = create(&store, &actor, K::Site, "Located site")?; + let host = child(&store, &actor, &site, K::Device, "Located host")?; + child(&store, &actor, &host, K::Interface, "Needle located")?; + let expected = create(&store, &actor, K::Interface, "Needle unplaced")?; + for index in 0..128 { + create(&store, &actor, K::Site, &format!("Unrelated {index}"))?; + } + let mut selector = graph_query(); + selector.kinds = vec![K::Interface as i32]; + selector.search = "Needle".into(); + selector.network.as_mut().ok_or("selection")?.unplaced = true; + let page = graph(&store, &actor, selector)?; + assert_eq!(page.total, 1); + assert_eq!(page.items[0].target.as_ref(), Some(&expected)); + Ok(()) +} + +// The same stable asset ID must not attach links from an earlier incarnation. +#[test] +fn topology_never_attaches_links_to_a_different_incarnation() -> TestResult { + use coordinode_core::graph::types::Value; + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let interface = create(&store, &actor, K::Interface, "Uplink")?; + let segment = create(&store, &actor, K::L2Domain, "Fabric")?; + call( + &store, + &actor, + C::Relation(api::Relationship { + relation: Some(wire::Relation { + from: Some(interface.clone()), + to: Some(segment.clone()), + kind: wire::RelationKind::On as i32, + ..Default::default() + }), + ..Default::default() + }), + )?; + assert_eq!( + graph(&store, &actor, graph_query())? + .topology + .ok_or("topology")? + .edges + .len(), + 1 + ); + let incarnation = omloom_contracts::ids::IncarnationId::generate().to_string(); + store.diagnostic_database().execute_cypher_shared( + "MATCH (n:OmRow {key:$key}) SET n.incarnation=$incarnation", + Some( + [ + ( + "key".into(), + Value::String(format!("asset/{}", segment.asset_id)), + ), + ("incarnation".into(), Value::String(incarnation.clone())), + ] + .into_iter() + .collect(), + ), + None, + None, + None, + )?; + let page = graph(&store, &actor, graph_query())?; + assert!( + page.items + .iter() + .any(|item| item.target.as_ref().is_some_and(|target| { + target.asset_id == segment.asset_id && target.incarnation == incarnation + })) + ); + assert!(page.topology.ok_or("topology")?.edges.is_empty()); + Ok(()) +} + +// Dense membership retains independent links and bounds the permitted output. +#[test] +fn topology_caps_dense_pages_and_keeps_parallel_links() -> TestResult { + use prost::Message; + let dir = tempfile::tempdir()?; + let mut store = Store::open(dir.path(), policy())?; + let actor = operator(); + let host = create(&store, &actor, K::Device, "Edge")?; + let port = child(&store, &actor, &host, K::Port, "Physical uplink")?; + let interface = child(&store, &actor, &port, K::Interface, "Logical uplink")?; + call( + &store, + &actor, + C::Relation(api::Relationship { + relation: Some(wire::Relation { + from: Some(port.clone()), + to: Some(interface.clone()), + kind: wire::RelationKind::Carries as i32, + ..Default::default() + }), + ..Default::default() + }), + )?; + let mut selection = graph_query(); + selection.roots = vec![port, interface]; + let parallel = graph(&store, &actor, selection.clone())? + .topology + .ok_or("topology")?; + assert_eq!(parallel.edges.len(), 2); + assert_ne!(parallel.edges[0].relation_id, parallel.edges[1].relation_id); + assert_eq!(parallel.edges[0].from, parallel.edges[1].from); + assert_eq!(parallel.edges[0].to, parallel.edges[1].to); + selection.roots.clear(); + selection.search = "Fabric".into(); + selection.limit = 25; + let mut ports = Vec::new(); + for index in 0..16 { + ports.push(child( + &store, + &actor, + &host, + K::Interface, + &format!("Fabric port {index:02}"), + )?); + } + let mut segments = Vec::new(); + for index in 0..9 { + segments.push(create( + &store, + &actor, + K::L2Domain, + &format!("Fabric segment {index}"), + )?); + } + let connect = |port: &wire::AssetRef, segment: &wire::AssetRef| { + call( + &store, + &actor, + C::Relation(api::Relationship { + relation: Some(wire::Relation { + from: Some(port.clone()), + to: Some(segment.clone()), + kind: wire::RelationKind::On as i32, + ..Default::default() + }), + ..Default::default() + }), + ) + }; + for port in &ports { + for segment in &segments { + connect(port, segment)?; + } + } + for index in 0..64 { + let off_page = create(&store, &actor, K::L2Domain, &format!("Outside {index:02}"))?; + connect(&ports[0], &off_page)?; + } + let before = store.usage()?; + store.set_diagnostic_query_check(|db, statement| { + if statement.contains("RETURN properties(r) AS p ORDER BY r.key LIMIT $limit") { + let plan = db + .explain_plan(statement, None) + .expect("topology plan") + .explain(); + assert!( + plan.contains("IndexScan(r:OmRow ON omloom_topology_pair(topology_pair))"), + "{plan}" + ); + } + }); + graph(&store, &actor, selection.clone())?; + store.set_diagnostic_query_check(|_, _| {}); + let mut times = Vec::new(); + let mut payload = 0; + for _ in 0..30 { + let start = std::time::Instant::now(); + let page = graph(&store, &actor, selection.clone())?; + times.push(start.elapsed()); + assert_eq!(page.total, 25); + payload = page.encoded_len(); + assert!(payload < 32 * 1024); + let topology = page.topology.ok_or("topology")?; + assert_eq!(topology.edges.len(), 128); + assert!(topology.more_edges); + assert_eq!( + topology + .edges + .iter() + .map(|edge| &edge.relation_id) + .collect::>() + .len(), + 128 + ); + } + times.sort_unstable(); + println!( + "Dense topology 25 nodes, 144 relations, {payload} bytes, 30 native debug reads: p95 {:?}, p99 {:?}", + times[28], times[29] + ); + assert_eq!(store.usage()?, before); + Ok(()) +} + +// Saved coordinates belong to exact visible incarnations and use the existing CAS. +#[test] +fn topology_layout_survives_reopen_and_filters_hidden_pins() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let segment = create(&store, &actor, K::L2Domain, "Segment")?; + let pin = api::NetworkPin { + target: Some(segment), + x: -20, + y: 43, + }; + let visible_pin = api::NetworkPin { + target: Some(create(&store, &actor, K::Vrf, "Visible VRF")?), + x: 100, + y: 50, + }; + let mut query = graph_query(); + query.network.as_mut().ok_or("network")?.layout = Some(api::NetworkLayout { + pins: vec![pin.clone(), visible_pin.clone()], + table: true, + }); + let R::Views(saved) = call( + &store, + &actor, + C::SaveView(api::SaveView { + view: Some(api::SavedView { + name: "Network work".into(), + query: Some(Box::new(query)), + audiences: vec!["operations".into()], + editors: vec!["operations".into()], + ..Default::default() + }), + publish: true, + ..Default::default() + }), + )? + else { + return Err("save result".into()); + }; + let original = saved.items.into_iter().next().ok_or("view")?; + drop(store); + let store = Store::open(dir.path(), policy())?; + let R::Views(reopened) = call(&store, &actor, C::Views(api::Empty {}))? else { + return Err("views".into()); + }; + assert_eq!(reopened.items, vec![original.clone()]); + + let mut editor = operator(); + editor.actor = "other-editor".into(); + editor.grants[0].scope = Scope::Kinds(vec![K::Vrf]); + let before = store.usage()?; + let R::Views(restricted) = call(&store, &editor, C::Views(api::Empty {}))? else { + return Err("views".into()); + }; + let layout = restricted.items[0] + .query + .as_ref() + .and_then(|q| q.network.as_ref()) + .and_then(|network| network.layout.as_ref()) + .ok_or("layout")?; + assert!(layout.table); + assert_eq!(layout.pins, vec![visible_pin.clone()]); + assert_eq!(store.usage()?, before); + call( + &store, + &editor, + C::SaveView(api::SaveView { + view: Some(restricted.items[0].clone()), + publish: true, + ..Default::default() + }), + )?; + assert_eq!(store.usage()?, before); + + let mut renamed = restricted.items[0].clone(); + renamed.name = "Renamed by restricted editor".into(); + let R::Views(result) = call( + &store, + &editor, + C::SaveView(api::SaveView { + view: Some(renamed), + publish: true, + ..Default::default() + }), + )? + else { + return Err("save result".into()); + }; + assert_eq!( + result.items[0] + .query + .as_ref() + .and_then(|q| q.network.as_ref()) + .and_then(|n| n.layout.as_ref()) + .ok_or("layout")? + .pins, + vec![visible_pin.clone()] + ); + let R::Views(author) = call(&store, &actor, C::Views(api::Empty {}))? else { + return Err("views".into()); + }; + assert_eq!( + author.items[0] + .query + .as_ref() + .and_then(|q| q.network.as_ref()) + .and_then(|n| n.layout.as_ref()) + .ok_or("layout")? + .pins, + vec![pin, visible_pin] + ); + + let mut changed = author.items[0].clone(); + changed.name = "First concurrent edit".into(); + call( + &store, + &actor, + C::SaveView(api::SaveView { + view: Some(changed), + publish: true, + ..Default::default() + }), + )?; + assert!(matches!( + call( + &store, + &editor, + C::SaveView(api::SaveView { + view: Some(original), + publish: true, + ..Default::default() + }) + ), + Err(Error::Conflict) + )); + Ok(()) +} + +// Deleting a definition only requires current authority over that definition. +#[test] +fn topology_owner_can_remove_a_view_with_an_unreadable_site() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let mut actor = operator(); + let site = create(&store, &actor, K::Site, "Site")?; + let mut query = graph_query(); + query.network.as_mut().ok_or("selection")?.sites.push(site); + let R::Views(saved) = call( + &store, + &actor, + C::SaveView(api::SaveView { + view: Some(api::SavedView { + name: "Site view".into(), + query: Some(Box::new(query)), + ..Default::default() + }), + ..Default::default() + }), + )? + else { + return Err("saved view".into()); + }; + actor.grants[0].scope = Scope::Kinds(vec![K::Interface]); + let result = call( + &store, + &actor, + C::SaveView(api::SaveView { + view: Some(saved.items[0].clone()), + remove: true, + ..Default::default() + }), + )?; + assert!(matches!(result, R::Mutation(_))); + let R::Views(remaining) = call(&store, &actor, C::Views(api::Empty {}))? else { + return Err("views".into()); + }; + assert!(remaining.items.is_empty()); + Ok(()) +} + +// Configured allocations are attached through Member rather than Relation. +#[test] +fn topology_projects_authorized_membership_assignments() -> TestResult { + let dir = tempfile::tempdir()?; + let store = Store::open(dir.path(), policy())?; + let actor = operator(); + let interface = create(&store, &actor, K::Interface, "Management")?; + let space = create(&store, &actor, K::AddressSpace, "Management space")?; + let R::Mutation(created) = call( + &store, + &actor, + C::Network(api::Network { + kind: K::IpAddress as i32, + name: "Management IP".into(), + cidr: "192.0.2.3/24".into(), + address_space: Some(space), + sharing: api::SharedUse::Exclusive as i32, + ..Default::default() + }), + )? + else { + return Err("address".into()); + }; + let address = created.targets[0].clone(); + call( + &store, + &actor, + C::Member(api::Member { + address: Some(address.clone()), + interface: Some(interface.clone()), + expected: store.asset(AssetRef::try_from(&address)?)?.revision, + ..Default::default() + }), + )?; + let before = store.usage()?; + let edge = graph(&store, &actor, graph_query())? + .topology + .ok_or("topology")? + .edges + .into_iter() + .find(|edge| edge.kind == wire::RelationKind::AssignedTo as i32) + .ok_or("assignment link missing")?; + assert_eq!(edge.from, Some(address)); + assert_eq!(edge.to, Some(interface)); + let mut reader = operator(); + reader.grants[0].read = Fields::Selected(vec!["name".into(), "relations".into()]); + assert!( + graph(&store, &reader, graph_query())? + .topology + .ok_or("topology")? + .edges + .is_empty() + ); + assert_eq!(store.usage()?, before); + Ok(()) +} diff --git a/crates/omloom-store/tests/workspace/views.rs b/crates/omloom-store/tests/workspace/views.rs index 5698c64..c974586 100644 --- a/crates/omloom-store/tests/workspace/views.rs +++ b/crates/omloom-store/tests/workspace/views.rs @@ -30,10 +30,10 @@ fn saved_views_validate_responsibility_filters_without_matching_assets() -> Test api::request::Command::SaveView(api::SaveView { view: Some(api::SavedView { name: "Owner view".into(), - query: Some(api::Query { + query: Some(Box::new(api::Query { responsibility: Some(filter), ..query() - }), + })), ..Default::default() }), ..Default::default() @@ -56,7 +56,7 @@ fn saved_view_admission_preserves_metadata_reply_budget() -> TestResult { let actor = operator(); let small = api::SavedView { name: "Small view".into(), - query: Some(query()), + query: Some(Box::new(query())), ..Default::default() }; let Reply::Views(saved) = call( @@ -88,7 +88,7 @@ fn saved_view_admission_preserves_metadata_reply_budget() -> TestResult { Command::SaveView(api::SaveView { view: Some(api::SavedView { name: format!("Large {index}"), - query: Some(large_query.clone()), + query: Some(Box::new(large_query.clone())), ..Default::default() }), ..Default::default() @@ -106,7 +106,7 @@ fn saved_view_admission_preserves_metadata_reply_budget() -> TestResult { } assert!(refused, "oversized saved-view collection admitted"); let mut expanded = small.clone(); - expanded.query = Some(large_query); + expanded.query = Some(Box::new(large_query)); let before = store.usage()?; assert!(matches!( call( @@ -390,7 +390,7 @@ fn personal_publication_and_shared_edit_conflicts() -> TestResult { api::request::Command::SaveView(api::SaveView { view: Some(api::SavedView { name: "Private topology".into(), - query: Some(query()), + query: Some(Box::new(query())), ..Default::default() }), ..Default::default() @@ -460,7 +460,7 @@ fn personal_publication_and_shared_edit_conflicts() -> TestResult { let api::reply::Result::Page(page) = call( &store, &colleague, - api::request::Command::Query(view.query.clone().expect("query")), + api::request::Command::Query(*view.query.clone().expect("query")), )? else { panic!("page") @@ -526,7 +526,7 @@ fn invalid_saved_predicate_is_not_silently_removed() -> TestResult { api::request::Command::SaveView(api::SaveView { view: Some(api::SavedView { name: "Backup hosts".into(), - query: Some(q.clone()), + query: Some(Box::new(q.clone())), ..Default::default() }), ..Default::default() @@ -553,7 +553,7 @@ fn invalid_saved_predicate_is_not_silently_removed() -> TestResult { panic!("views") }; assert_eq!(views.items[0].id, saved.items[0].id); - assert_eq!(views.items[0].query.as_ref(), Some(&q)); + assert_eq!(views.items[0].query.as_deref(), Some(&q)); assert!(matches!( call(&store, &context, api::request::Command::Query(q)), Err(Error::Incomplete) diff --git a/docs/inventory.md b/docs/inventory.md index 65a0d5c..0b98d5f 100644 --- a/docs/inventory.md +++ b/docs/inventory.md @@ -392,6 +392,44 @@ IPAM selection uses its namespace and address-range semantics independently of physical ownership. Hidden ownership does not conceal otherwise permitted resources or grant access to their containers. +### Network topology + +The topology index is part of the initial storage schema. Use a fresh store +when trying this feature with a development database created before that index +was introduced. + +Network opens a diagram of known current resources. Switch to **Table** for the +same selection with columns, bulk actions and ordinary list navigation. **Site** +and **Project / selection** are independent. Choose one or more permitted sites, +or **No known site**, and intersect that choice with a saved resource selection. +An empty result keeps both filters in place. + +Open **Resource navigator** to browse entities by ownership. Expanding a row +fetches that level and leaves other branches open. Selecting a container scopes +both the diagram and the table to its network branch, including nested resources, +while retaining the site and project filters. Selecting a segment opens its +details and paged L2 participants, including VMs whose interface is still unknown. + +The diagram draws relationships whose two endpoints are on the current page. +Open a resource to see additional authorized relationship pages and use the +ordinary inventory editor. The recorded link type describes the evidence. +Missing links, partial observations and disconnected cards do not establish +network reachability. On a phone, opening a card leads to the full detail +screen with a return action. + +**Arrange** enables dragging or arrow-key movement of cards. **Save view** +stores only explicitly arranged positions and the chosen presentation. +Refreshing, zooming and rearranging before saving leave inventory untouched. +Shared definitions use their existing audience and editor rules, and the store +rechecks access to pinned resource incarnations when loading them. + +The browser requests 25 nodes per page. A network projection returns at most +128 recorded links and uses the existing response byte limit. Site selection +accepts up to 16 explicit sites, and traversal from those sites uses the common +1,024-resource scope ceiling. **No known site** checks visible ownership above +the matching resources. Narrow the selection if it exceeds 4,096 candidates +or the ownership traversal exceeds its work limit. + List context is limited to authorized owner labels and address hints for the displayed services/interfaces. It does not load host routing tables, firewall rules or the address sets of ancestor hosts and sibling services. Network diff --git a/proto/omloom/v1/workspace.proto b/proto/omloom/v1/workspace.proto index e1f1887..7c757b2 100644 --- a/proto/omloom/v1/workspace.proto +++ b/proto/omloom/v1/workspace.proto @@ -75,8 +75,9 @@ message Query { uint32 limit = 10; // Explicit selection of objects without a confirmed physical placement. bool unplaced = 11; - // Explicit resource perspective within roots, rechecked by the server. Only - // its local components are projected; upstream topology is not inferred. + // Explicit resource perspective within roots, rechecked by the server. + // In Network, level_only=false selects the bounded ownership branch. + // Other focused queries project local components. optional omloom.v1.AssetRef focus = 12; // Request bounded owner/address context for visible rows, not additional authority. bool include_context = 13; @@ -91,6 +92,19 @@ message Query { // Exact issuer-qualified effective responsibility filter, applied before totals // and pagination. Only a permitted facet can contribute a matching resource. optional ResponsibilityFilter responsibility = 16; + // Independent location selector and optional current Network projection. + optional NetworkSelection network = 18; +} +message NetworkSelection { + // Explicit permitted site roots, intersected with the independent resource + // selection. Empty includes every location, including unknown location. + repeated omloom.v1.AssetRef sites = 1; + // Request typed edges within this page of the Network view. + bool include_topology = 2; + // Explicit Network presentation preferences. Used only by saved definitions. + optional NetworkLayout layout = 3; + // Select resources without a known permitted site. Mutually exclusive with sites. + bool unplaced = 4; } message ResponsibilityFilter { // Independent owner/operator facet selected by the user; never a permission selector. @@ -504,6 +518,24 @@ message Page { uint64 revision = 4; // Bounded authorized owner/address hints for this page only, absent unless requested. repeated ItemContext contexts = 5; + // Current authorized edges between emitted nodes, present only when requested. + optional NetworkTopology topology = 6; +} +message NetworkTopology { + // Permitted typed links, with independent identities for parallel links. + repeated NetworkLink edges = 1; + // True when the page has more permitted internal edges than the drawing limit. + bool more_edges = 2; +} +message NetworkLink { + // Stable identity of the recorded relation, used to distinguish parallel links. + string relation_id = 1; + // Permitted source incarnation on this page, as recorded by the relationship. + omloom.v1.AssetRef from = 2; + // Permitted destination incarnation on this page, as recorded by the relationship. + omloom.v1.AssetRef to = 3; + // Type of recorded evidence, which does not imply packet reachability. + omloom.v1.RelationKind kind = 4; } message OwnerContext { // Current authorized owner identity; never a fabricated unknown resource. @@ -934,6 +966,20 @@ message SavedView { // Explicit team editors, constrained to the publication audience. repeated string editors = 7; } +message NetworkLayout { + // Explicit table preference. Omission uses the topology default. + bool table = 1; + // User-arranged positions, at most 64, for exact permitted incarnations. + repeated NetworkPin pins = 2; +} +message NetworkPin { + // Exact resource positioned by the user, reauthorized when opening a view. + omloom.v1.AssetRef target = 1; + // Horizontal position relative to the diagram origin, in units, ±4096. + sint32 x = 2; + // Vertical position relative to the diagram origin, in units, ±4096. + sint32 y = 3; +} message SaveView { // Definition the user edited; server ignores/rejects caller-supplied owner changes. SavedView view = 1; diff --git a/scripts/install-ci-release-tool.sh b/scripts/install-ci-release-tool.sh new file mode 100644 index 0000000..b56e91e --- /dev/null +++ b/scripts/install-ci-release-tool.sh @@ -0,0 +1,44 @@ +#!/bin/sh +set -eu + +tool=${1:?Specify mold or protoc} +case "$tool:$(uname -s):$(uname -m)" in + mold:Linux:x86_64) + repo=rui314/mold + asset=612122001 # mold-3.0.0-x86_64-linux.tar.gz + checksum=6c90d4a474c7c0409dfb575be03a5345878ac14fdba18de8b40fa58c60121189 + ;; + protoc:Linux:x86_64) + repo=protocolbuffers/protobuf + asset=570742434 # protoc-36.2-linux-x86_64.zip + checksum=121f6c7afe1d4d0e3ea6aab9432038599250134cbf4474cb1167d2c7decd4278 + ;; + protoc:Darwin:arm64) + repo=protocolbuffers/protobuf + asset=570742436 # protoc-36.2-osx-aarch_64.zip + checksum=9cd98a532c5c5e0c4161314de0225de27e4c8a323917b6ea7b1b714d3ae23466 + ;; + protoc:Darwin:x86_64) + repo=protocolbuffers/protobuf + asset=570742448 # protoc-36.2-osx-x86_64.zip + checksum=228cc7add4616cc14ca5e80dee83209d44449a7aee95a914ae748fa374efb078 + ;; + *) + printf 'Unsupported CI tool or platform: %s\n' "$tool" >&2 + exit 2 + ;; +esac + +destination=$(mktemp -d "${RUNNER_TEMP:?}/omloom-$tool.XXXXXX") +archive=$(mktemp "${RUNNER_TEMP}/omloom-release.XXXXXX") +trap 'rm -f "$archive"' EXIT + +# Use the release asset API because browser download URLs can fail at the GitHub edge. +gh api -H 'Accept: application/octet-stream' "repos/$repo/releases/assets/$asset" > "$archive" +printf '%s %s\n' "$checksum" "$archive" | shasum -a 256 -c - +case "$tool" in + mold) tar --strip-components=1 -xzf "$archive" -C "$destination" ;; + protoc) unzip -q "$archive" -d "$destination" ;; +esac +"$destination/bin/$tool" --version +printf '%s\n' "$destination/bin" >> "${GITHUB_PATH:?}" diff --git a/web/app/App.vue b/web/app/App.vue index 419eae8..ecb4e9b 100644 --- a/web/app/App.vue +++ b/web/app/App.vue @@ -1,5 +1,6 @@