diff --git a/docs/explanation/tenant-types.md b/docs/explanation/tenant-types.md index d3faad0a..ac818c19 100644 --- a/docs/explanation/tenant-types.md +++ b/docs/explanation/tenant-types.md @@ -71,6 +71,17 @@ own; the client only needs enough types to turn JSON into bytes and back. - A library that another library depends on, or whose types entity schemas use, cannot be deleted, even with `mode=FORCE`. The schemas show up as references to the library, so impact analysis lists them too. +- An annotation in a `metadata` list, including the `metadata` of fields + nested in a built-in schema, is checked against the built-in types and the + live libraries in the entity's namespace, counting libraries written + earlier in the same batch. A type URL neither declares raises the + `UNKNOWN_ANNOTATION_TYPE` warning, with the annotation's path in + `subject_field`. The write still succeeds, so annotations keep working + while their library is drafted. Built-in annotations match by their full + message name under any host, so a look-alike such as + `acme.v1.MyOrphanAnnotation` suppresses nothing, and a misspelled one says + so instead of failing silently. If the namespace's libraries fail to compile, the warning still names every + type URL that is not built in and says the libraries could not be checked. ## Breaking changes become new packages diff --git a/rsworkspace/crates/trogon-atlas-server/fuzz/Cargo.lock b/rsworkspace/crates/trogon-atlas-server/fuzz/Cargo.lock index c661624a..480308e6 100644 --- a/rsworkspace/crates/trogon-atlas-server/fuzz/Cargo.lock +++ b/rsworkspace/crates/trogon-atlas-server/fuzz/Cargo.lock @@ -947,7 +947,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1853,7 +1853,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2337,7 +2337,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2688,7 +2688,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3367,7 +3367,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index c9aa3a0b..6e74084b 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -890,10 +890,10 @@ fn swimlane_state_endpoint( } pub(crate) fn is_derived_field(field: &pb::FieldSpec) -> bool { - field.metadata.iter().any(|m| { - m.type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.DerivedFieldAnnotation") - }) + field + .metadata + .iter() + .any(|m| crate::annotation_type::BuiltinAnnotation::DERIVED_FIELD.matches(m)) } fn fields_of(entity: &pb::Entity) -> Vec { diff --git a/rsworkspace/crates/trogon-atlas-server/src/annotation_type.rs b/rsworkspace/crates/trogon-atlas-server/src/annotation_type.rs new file mode 100644 index 00000000..8f8e1bfc --- /dev/null +++ b/rsworkspace/crates/trogon-atlas-server/src/annotation_type.rs @@ -0,0 +1,64 @@ +use trogon_atlas_types::TypeUrl; + +/// A built-in annotation message, recognized only by its exact protobuf +/// full name so a look-alike type from another package never passes for it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct BuiltinAnnotation(&'static str); + +impl BuiltinAnnotation { + pub(crate) const ASSUMPTION: Self = + Self("trogonatlas.annotation.v1alpha1.AssumptionAnnotation"); + pub(crate) const CODE_REF: Self = Self("trogonatlas.annotation.v1alpha1.CodeRefAnnotation"); + pub(crate) const DERIVED_FIELD: Self = + Self("trogonatlas.eventmodel.v1alpha1.DerivedFieldAnnotation"); + pub(crate) const LIFECYCLE: Self = Self("trogonatlas.annotation.v1alpha1.LifecycleAnnotation"); + pub(crate) const MIXED_SURFACE: Self = + Self("trogonatlas.eventmodel.v1alpha1.MixedSurfaceAnnotation"); + pub(crate) const OPEN_QUESTION: Self = + Self("trogonatlas.annotation.v1alpha1.OpenQuestionAnnotation"); + pub(crate) const ORPHAN: Self = Self("trogonatlas.eventmodel.v1alpha1.OrphanAnnotation"); + pub(crate) const PERSONAL_DATA: Self = + Self("trogonatlas.eventmodel.v1alpha1.PersonalDataAnnotation"); + pub(crate) const PINNED_REF: Self = Self("trogonatlas.eventmodel.v1alpha1.PinnedRefAnnotation"); + pub(crate) const SHARED_CONSUMERS: Self = + Self("trogonatlas.eventmodel.v1alpha1.SharedConsumersAnnotation"); + pub(crate) const UUIDV5_IDENTITY: Self = + Self("trogonatlas.eventmodel.v1alpha1.Uuidv5IdentityAnnotation"); + + pub(crate) fn matches(self, any: &prost_types::Any) -> bool { + TypeUrl::parse(&any.type_url).is_ok_and(|url| url.full_name() == self.0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn any(type_url: &str) -> prost_types::Any { + prost_types::Any { + type_url: type_url.into(), + value: Vec::new(), + } + } + + #[test] + fn matches_the_exact_message_under_any_host() { + for url in [ + "type.googleapis.com/trogonatlas.eventmodel.v1alpha1.OrphanAnnotation", + "example.com/types/trogonatlas.eventmodel.v1alpha1.OrphanAnnotation", + ] { + assert!(BuiltinAnnotation::ORPHAN.matches(&any(url)), "{url}"); + } + } + + #[test] + fn rejects_names_that_only_share_a_suffix() { + for url in [ + "type.googleapis.com/acme.trogonatlas.eventmodel.v1alpha1.OrphanAnnotation", + "type.googleapis.com/acmetrogonatlas.eventmodel.v1alpha1.OrphanAnnotation", + "trogonatlas.eventmodel.v1alpha1.OrphanAnnotation", + ] { + assert!(!BuiltinAnnotation::ORPHAN.matches(&any(url)), "{url}"); + } + } +} diff --git a/rsworkspace/crates/trogon-atlas-server/src/lib.rs b/rsworkspace/crates/trogon-atlas-server/src/lib.rs index d6bce346..cab66a6d 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/lib.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/lib.rs @@ -1,4 +1,5 @@ pub mod analysis; +mod annotation_type; pub mod artifacts; pub mod as_of; pub mod assessment; diff --git a/rsworkspace/crates/trogon-atlas-server/src/service.rs b/rsworkspace/crates/trogon-atlas-server/src/service.rs index c145cd1a..018634a5 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/service.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/service.rs @@ -655,10 +655,7 @@ impl Drop for StreamChangesGuard { fn entity_lifecycle_status(entity: &pb::Entity) -> Option { let metadata = crate::validation::entity_metadata(entity); for any in metadata { - if !any - .type_url - .ends_with("trogonatlas.annotation.v1alpha1.LifecycleAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::LIFECYCLE.matches(any) { continue; } if let Ok(ann) = ::decode(any.value.as_slice()) { @@ -7395,6 +7392,16 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl .type_libraries .check_schema(&*self.store, &entity, &[], branch.as_deref()) .await?; + let annotation_issues = validation::unknown_annotation_type_issues( + &pb::EntityRef { + kind: kind as i32, + id: Some(id.clone()), + }, + &self + .type_libraries + .unknown_annotation_types(&*self.store, &entity, &[], branch.as_deref()) + .await?, + ); let ran_scenarios = runs_scenarios(kind); let ran_strategic = is_strategic_kind(kind); @@ -7416,6 +7423,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl &scenario_issues, ))); } + scenario_issues.extend(annotation_issues); // Strategic findings never block: sequential puts of one problem // space are order dependent, so a context put before its subdomains // is transiently dangling and rejecting it would make the model @@ -8436,18 +8444,38 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl _ => None, }) .collect(); - for m in metas.iter().filter(|m| m.is_put) { - if let Some(entity) = m.entity.as_ref() { - if let Some(admission) = self - .type_libraries - .check_schema(&*self.store, entity, &pending_libraries, branch.as_deref()) - .await? - { - if !type_library_admissions.contains(&admission) { - type_library_admissions.push(admission); - } + let mut annotation_issues: Vec> = + std::iter::repeat_with(Vec::new).take(metas.len()).collect(); + let removed_libraries: Vec = metas + .iter() + .filter(|m| !m.is_put && m.kind == pb::EntityKind::TypeLibrary) + .map(|m| m.id.clone()) + .collect(); + let mut annotation_types = self.type_libraries.annotation_lookup( + &pending_libraries, + &removed_libraries, + branch.as_deref(), + ); + for (m, issues) in metas.iter().zip(&mut annotation_issues) { + let Some(entity) = m.entity.as_ref().filter(|_| m.is_put) else { + continue; + }; + if let Some(admission) = self + .type_libraries + .check_schema(&*self.store, entity, &pending_libraries, branch.as_deref()) + .await? + { + if !type_library_admissions.contains(&admission) { + type_library_admissions.push(admission); } } + *issues = validation::unknown_annotation_type_issues( + &pb::EntityRef { + kind: m.kind as i32, + id: Some(m.id.clone()), + }, + &annotation_types.unknown(&*self.store, entity).await?, + ); } let invariant_changes: Vec = metas @@ -8607,8 +8635,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl // replacement semantics, so the order of ops inside a batch never // matters. Each finding rides the op whose entity it is about, or // the op that caused it when that entity is not in the batch. - let mut per_op_validation: Vec> = - std::iter::repeat_with(Vec::new).take(metas.len()).collect(); + let mut per_op_validation = annotation_issues; let has_strategic_ops = has_strategic_puts || has_strategic_deletes; let architecture_subjects = metas .iter() @@ -8651,7 +8678,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl &changes, &base, &world, )) { - per_op_validation[*op_index] = issues; + per_op_validation[*op_index].extend(issues); } let delete_ops: Vec = metas .iter() @@ -8674,7 +8701,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl &deletions, &before, &world, &reported, )) { - per_op_validation[*op_index] = issues; + per_op_validation[*op_index].extend(issues); } } } @@ -8710,7 +8737,7 @@ impl pb::event_model_service_server::EventModelService for EventModelServiceImpl .into_iter() .partition(|iss| iss.severity == pb::validation_issue::Severity::Error as i32); if errors.is_empty() { - per_op_validation[i] = non_errors; + per_op_validation[i].extend(non_errors); } else { if first_error_index.is_none() { first_error_index = Some(i); diff --git a/rsworkspace/crates/trogon-atlas-server/src/type_libraries.rs b/rsworkspace/crates/trogon-atlas-server/src/type_libraries.rs index deb685b0..e81edb64 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/type_libraries.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/type_libraries.rs @@ -431,7 +431,7 @@ impl TypeLibraryGate { type_url: &TypeUrl, branch: Option<&str>, ) -> Result { - let live = self.live_pool(store, namespace, &[], branch).await?; + let live = self.live_pool_status(store, namespace, &[], branch).await?; let message = live .pool .get_message_by_name(type_url.full_name()) @@ -458,7 +458,7 @@ impl TypeLibraryGate { namespace: &str, branch: Option<&str>, ) -> Result { - let live = self.live_pool(store, namespace, &[], branch).await?; + let live = self.live_pool_status(store, namespace, &[], branch).await?; let files: Vec = live .pool .files() @@ -494,7 +494,9 @@ impl TypeLibraryGate { pending: &[&pb::TypeLibrary], branch: Option<&str>, ) -> Result, Status> { - let live = self.live_pool(store, namespace, pending, branch).await?; + let live = self + .live_pool_status(store, namespace, pending, branch) + .await?; let files: Vec = live .pool .files() @@ -533,7 +535,7 @@ impl TypeLibraryGate { let type_url = TypeUrl::parse(&schema.type_url) .map_err(|e| invalid(format!("schema.type_url: {e}")))?; let live = self - .live_pool(store, &id.namespace, pending, branch) + .live_pool_status(store, &id.namespace, pending, branch) .await?; let message = live .pool @@ -563,6 +565,69 @@ impl TypeLibraryGate { Ok(Some(live.admission)) } + /// Annotations on `entity`, at any depth, whose type neither the + /// built-in types nor the namespace's live type libraries declare. + /// `pending` are type libraries written alongside the entity. Live + /// libraries that do not compile together, and a compiler that times out + /// or is unavailable, leave tenant types unresolved instead of refusing + /// the write. A failure to load the libraries still refuses it. + /// + /// # Errors + /// + /// Store failures while loading the namespace's type libraries. + pub async fn unknown_annotation_types( + &self, + store: &dyn Store, + entity: &pb::Entity, + pending: &[&pb::TypeLibrary], + branch: Option<&str>, + ) -> Result { + let builtins = self.pools.builtins(); + let unknown = annotations_outside(builtins, entity); + let Some(id) = + trogon_atlas_core::refs::entity_id(entity).filter(|_| unknown.needs_libraries()) + else { + return Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::Checked, + }); + }; + match self + .live_pool(store, &id.namespace, pending, &[], branch) + .await + { + Ok(live) => Ok(UnknownAnnotationTypes { + annotations: annotations_outside(&live.pool, entity).annotations, + libraries: LibraryConsultation::Checked, + }), + Err(PoolLookup::DidNotCompile(_)) => Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::DidNotCompile, + }), + Err(PoolLookup::CompilerUnavailable(_)) => Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::CompilerUnavailable, + }), + Err(PoolLookup::Propagate(status)) => Err(status), + } + } + + /// One type-library lookup per namespace, shared by every put in a batch. + pub(crate) fn annotation_lookup<'a>( + &'a self, + pending: &'a [&'a pb::TypeLibrary], + removed: &'a [pb::Id], + branch: Option<&'a str>, + ) -> AnnotationTypeLookup<'a> { + AnnotationTypeLookup { + gate: self, + pending, + removed, + branch, + pools: HashMap::new(), + } + } + fn is_builtin(&self, file: &FileDescriptor) -> bool { self.pools .builtins() @@ -570,14 +635,34 @@ impl TypeLibraryGate { .is_some() } - async fn live_pool( + async fn live_pool_status( &self, store: &dyn Store, namespace: &str, pending: &[&pb::TypeLibrary], branch: Option<&str>, ) -> Result { - let mut registered = load(store, namespace, branch).await?; + self.live_pool(store, namespace, pending, &[], branch) + .await + .map_err(PoolLookup::status) + } + + async fn live_pool( + &self, + store: &dyn Store, + namespace: &str, + pending: &[&pb::TypeLibrary], + removed: &[pb::Id], + branch: Option<&str>, + ) -> Result { + let mut registered = load(store, namespace, branch) + .await + .map_err(PoolLookup::Propagate)?; + for id in removed { + if id.namespace == namespace { + registered.remove(&LibraryVersion::of(id)); + } + } let admission = TypeLibraryAdmission { namespace: namespace.to_owned(), seen: registered @@ -603,9 +688,12 @@ impl TypeLibraryGate { let Some(id) = entry.library.id.as_ref() else { continue; }; - let source = self.source_of(&entry.library)?; + let source = self + .source_of(&entry.library) + .map_err(PoolLookup::Propagate)?; owners.push((source.prefix().clone(), id.clone())); - let dependencies = dependency_keys(id, &entry.library)? + let dependencies = dependency_keys(id, &entry.library) + .map_err(PoolLookup::Propagate)? .into_iter() .map(|dependency| { live.get(dependency.slug.as_str()) @@ -622,7 +710,7 @@ impl TypeLibraryGate { .pools .get_or_build(PoolScope::new(namespace, branch), members) .await - .map_err(|e| pool_err(namespace, e))?; + .map_err(|e| pool_lookup(namespace, e))?; Ok(LivePool { pool, owners, @@ -747,6 +835,244 @@ fn unknown_in_value(value: &Value) -> Option { } } +/// An annotation whose type url names no message the namespace can resolve. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnknownAnnotation { + pub type_url: String, + /// Structural path to the annotation inside the entity kind, e.g. + /// `metadata[0]` or `schema.fields[2].metadata[1]`. + pub path: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnknownAnnotationTypes { + pub annotations: Vec, + /// Whether the namespace's type libraries were consulted. + pub libraries: LibraryConsultation, +} + +/// Why tenant types were or were not part of an annotation check. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LibraryConsultation { + /// The live libraries compiled and were consulted. + Checked, + /// The libraries were loaded and do not compile together. + DidNotCompile, + /// The compiler timed out or was unavailable, so it never judged the libraries. + CompilerUnavailable, +} + +enum CachedPool { + Ready(Arc), + DidNotCompile, + CompilerUnavailable, +} + +pub(crate) struct AnnotationTypeLookup<'a> { + gate: &'a TypeLibraryGate, + pending: &'a [&'a pb::TypeLibrary], + removed: &'a [pb::Id], + branch: Option<&'a str>, + pools: HashMap, +} + +impl AnnotationTypeLookup<'_> { + pub(crate) async fn unknown( + &mut self, + store: &dyn Store, + entity: &pb::Entity, + ) -> Result { + let builtins = self.gate.pools.builtins(); + let unknown = annotations_outside(builtins, entity); + let Some(id) = + trogon_atlas_core::refs::entity_id(entity).filter(|_| unknown.needs_libraries()) + else { + return Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::Checked, + }); + }; + if !self.pools.contains_key(&id.namespace) { + let cached = match self + .gate + .live_pool(store, &id.namespace, self.pending, self.removed, self.branch) + .await + { + Ok(live) => CachedPool::Ready(live.pool), + Err(PoolLookup::DidNotCompile(_)) => CachedPool::DidNotCompile, + Err(PoolLookup::CompilerUnavailable(_)) => CachedPool::CompilerUnavailable, + Err(PoolLookup::Propagate(status)) => return Err(status), + }; + self.pools.insert(id.namespace.clone(), cached); + } + match self.pools.get(&id.namespace) { + Some(CachedPool::Ready(pool)) => Ok(UnknownAnnotationTypes { + annotations: annotations_outside(pool, entity).annotations, + libraries: LibraryConsultation::Checked, + }), + Some(CachedPool::DidNotCompile) => Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::DidNotCompile, + }), + Some(CachedPool::CompilerUnavailable) => Ok(UnknownAnnotationTypes { + annotations: unknown.annotations, + libraries: LibraryConsultation::CompilerUnavailable, + }), + None => Err(Status::internal("annotation type pool was not recorded")), + } + } +} + +const ANY_MESSAGE: &str = "google.protobuf.Any"; +const ENTITY_MESSAGE: &str = "trogonatlas.eventmodel.v1alpha1.Entity"; + +struct AnnotationScan { + annotations: Vec, + /// An `Any` named a message this pool does not have, so its contents + /// were not walked. A later pool may be able to. + opaque_payload: bool, +} + +impl AnnotationScan { + fn empty() -> Self { + Self { + annotations: Vec::new(), + opaque_payload: false, + } + } + + fn needs_libraries(&self) -> bool { + !self.annotations.is_empty() || self.opaque_payload + } +} + +/// Every `metadata` annotation in `entity` whose type url `pool` cannot +/// resolve, walking into nested messages and resolvable Any payloads. +fn annotations_outside(pool: &DescriptorPool, entity: &pb::Entity) -> AnnotationScan { + let Some(descriptor) = pool.get_message_by_name(ENTITY_MESSAGE) else { + return AnnotationScan::empty(); + }; + let Ok(mut message) = DynamicMessage::decode(descriptor, entity.encode_to_vec().as_slice()) + else { + return AnnotationScan::empty(); + }; + let mut pending: Vec<(Value, String)> = message + .take_fields() + .map(|(_, value)| (value, String::new())) + .collect(); + let mut out = AnnotationScan::empty(); + while let Some((value, path)) = pending.pop() { + match value { + Value::Message(mut message) => { + if message.descriptor().full_name() == ANY_MESSAGE { + if let Some(payload) = any_payload(pool, &mut message) { + pending.push((Value::Message(payload), path)); + } else if names_unknown_message(pool, &message) { + out.opaque_payload = true; + } + continue; + } + for (field, value) in message.take_fields() { + let child = if path.is_empty() { + field.name().to_owned() + } else { + format!("{path}.{}", field.name()) + }; + if field.name() == "metadata" { + if let Value::List(values) = &value { + for (i, entry) in values.iter().enumerate() { + if let Value::Message(any) = entry { + if let Some(type_url) = unresolved_type_url(pool, any) { + out.annotations.push(UnknownAnnotation { + type_url, + path: format!("{child}[{i}]"), + }); + } + } + } + } + } + pending.push((value, child)); + } + } + Value::List(values) => pending.extend( + values + .into_iter() + .enumerate() + .map(|(i, value)| (value, format!("{path}[{i}]"))), + ), + Value::Map(values) => { + pending.extend( + values + .into_iter() + .map(|(key, value)| (value, format!("{path}[{}]", map_key(&key)))), + ); + } + _ => {} + } + } + out.annotations.sort_by(|a, b| a.path.cmp(&b.path)); + out +} + +fn names_unknown_message(pool: &DescriptorPool, any: &DynamicMessage) -> bool { + let Some(raw) = any_type_url(any) else { + return false; + }; + TypeUrl::parse(&raw).is_ok_and(|url| pool.get_message_by_name(url.full_name()).is_none()) +} + +fn map_key(key: &prost_reflect::MapKey) -> String { + match key { + prost_reflect::MapKey::Bool(value) => value.to_string(), + prost_reflect::MapKey::I32(value) => value.to_string(), + prost_reflect::MapKey::I64(value) => value.to_string(), + prost_reflect::MapKey::U32(value) => value.to_string(), + prost_reflect::MapKey::U64(value) => value.to_string(), + prost_reflect::MapKey::String(value) => value.clone(), + } +} + +fn any_type_url(any: &DynamicMessage) -> Option { + match any.get_field_by_name("type_url")?.as_ref() { + Value::String(url) => Some(url.clone()), + _ => None, + } +} + +fn unresolved_type_url(pool: &DescriptorPool, any: &DynamicMessage) -> Option { + if any.descriptor().full_name() != ANY_MESSAGE { + return None; + } + let raw = any_type_url(any).unwrap_or_default(); + let known = TypeUrl::parse(&raw).is_ok_and(|url| annotation_type_is_known(pool, &url)); + (!known).then_some(raw) +} + +fn annotation_type_is_known(pool: &DescriptorPool, url: &TypeUrl) -> bool { + let Some(message) = pool.get_message_by_name(url.full_name()) else { + return false; + }; + let parent = message.parent_file(); + let file = parent.name(); + if file.starts_with("google/protobuf/") { + return false; + } + if file.starts_with("trogonatlas/") { + return message.name().ends_with("Annotation"); + } + true +} + +fn any_payload(pool: &DescriptorPool, any: &mut DynamicMessage) -> Option { + let url = TypeUrl::parse(&any_type_url(any)?).ok()?; + let descriptor = pool.get_message_by_name(url.full_name())?; + let Some(Value::Bytes(bytes)) = any.take_field_by_name("value") else { + return None; + }; + DynamicMessage::decode(descriptor, bytes).ok() +} + fn unknown_type(namespace: &str, type_url: &TypeUrl) -> Status { Status::with_error_details( Code::NotFound, @@ -758,15 +1084,35 @@ fn unknown_type(namespace: &str, type_url: &TypeUrl) -> Status { ) } -fn pool_err(namespace: &str, error: PoolError) -> Status { +enum PoolLookup { + /// The libraries were loaded and do not compile together. + DidNotCompile(Status), + /// The compiler timed out or was unavailable, so it never judged them. + CompilerUnavailable(Status), + /// The libraries could not be loaded, or one of them is not a valid + /// library. The write cannot proceed. + Propagate(Status), +} + +impl PoolLookup { + fn status(self) -> Status { + match self { + Self::DidNotCompile(status) + | Self::CompilerUnavailable(status) + | Self::Propagate(status) => status, + } + } +} + +fn pool_lookup(namespace: &str, error: PoolError) -> PoolLookup { match error { PoolError::Compile { error: error @ (CompileError::TimedOut(_) | CompileError::Unavailable), .. - } => compile_err(error), - other => Status::failed_precondition(format!( + } => PoolLookup::CompilerUnavailable(compile_err(error)), + other => PoolLookup::DidNotCompile(Status::failed_precondition(format!( "the live type libraries in namespace {namespace} do not compile together: {other}" - )), + ))), } } diff --git a/rsworkspace/crates/trogon-atlas-server/src/validation.rs b/rsworkspace/crates/trogon-atlas-server/src/validation.rs index 6b8baad0..25fd62c1 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/validation.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/validation.rs @@ -556,6 +556,9 @@ pub fn rule_catalog() -> Vec { r("SCHEMA_TYPE_URL_EMPTY", Error, "Schema Any has no type_url", "An Any-wrapped Schema must carry its type_url so consumers can deserialize it.", Unspecified, "schema", "schema"), + r("UNKNOWN_ANNOTATION_TYPE", Warning, "Annotation type is not known", + "A metadata annotation's type_url names no built-in annotation and no message the namespace's live type libraries declare, so no reader can decode it and any rule keyed on it never sees it. Usually a misspelled name or a missing type library. Reported when the entity is written; the write still succeeds.", + Unspecified, "metadata", "schema"), // ----- personal data (PersonalDataAnnotation) ----------------------------- r("PII_DOC_MISSING", Error, "PersonalDataAnnotation has no doc, or is malformed", "A PersonalDataAnnotation's `doc` is empty or only whitespace, or the Any payload cannot be decoded as a PersonalDataAnnotation at all. Either way the annotation says nothing: explain what personal data this is and whose.", @@ -754,6 +757,7 @@ mod catalog_drift_tests { "SCENARIO_REJECT_NO_REASON", "SCENARIO_TRANSITION_VIOLATION", "SCHEMA_TYPE_URL_EMPTY", + "UNKNOWN_ANNOTATION_TYPE", "SCREEN_MANY_CONTRIBUTORS", "SCREEN_NAMESPACE_IS_CONTEXT", "SEAM_WITHOUT_RELATIONSHIP", @@ -1000,10 +1004,7 @@ fn push_annotation_findings( issues: &mut Vec, ) { for any in metadata { - if any - .type_url - .ends_with("trogonatlas.annotation.v1alpha1.OpenQuestionAnnotation") - { + if crate::annotation_type::BuiltinAnnotation::OPEN_QUESTION.matches(any) { if let Ok(a) = ::decode(any.value.as_slice()) { @@ -1019,10 +1020,7 @@ fn push_annotation_findings( ..Default::default() }); } - } else if any - .type_url - .ends_with("trogonatlas.annotation.v1alpha1.AssumptionAnnotation") - { + } else if crate::annotation_type::BuiltinAnnotation::ASSUMPTION.matches(any) { if let Ok(a) = ::decode(any.value.as_slice()) { @@ -1093,10 +1091,7 @@ pub(crate) fn is_derived_field(field: &pb::FieldSpec) -> bool { // matching would let an empty-doc annotation silently suppress // EVENT_FIELD_UNSOURCED / AUTOMATION_COMMAND_FIELD_UNSOURCED. for any in &field.metadata { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.DerivedFieldAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::DERIVED_FIELD.matches(any) { continue; } if let Ok(ann) = @@ -1180,6 +1175,73 @@ fn walk_personal_data_fields<'a>( } } +/// `UNKNOWN_ANNOTATION_TYPE` for each annotation on `subject` whose type the +/// namespace cannot resolve. +#[must_use] +pub fn unknown_annotation_type_issues( + subject: &pb::EntityRef, + found: &crate::type_libraries::UnknownAnnotationTypes, +) -> Vec { + let cause = match found.libraries { + crate::type_libraries::LibraryConsultation::Checked => { + "neither a built-in annotation nor any live type library in the namespace declares it" + } + crate::type_libraries::LibraryConsultation::DidNotCompile => { + "no built-in annotation declares it, and the namespace's type libraries do not compile together so tenant types could not be checked" + } + crate::type_libraries::LibraryConsultation::CompilerUnavailable => { + "no built-in annotation declares it, and the type compiler was unavailable so tenant types could not be checked" + } + }; + let mut issues: Vec = found + .annotations + .iter() + .map(|annotation| pb::ValidationIssue { + severity: pb::validation_issue::Severity::Warning as i32, + code: "UNKNOWN_ANNOTATION_TYPE".into(), + message: format!( + "annotation type {:?} is unknown: {cause}", + annotation.type_url + ), + subject: Some(subject.clone()), + subject_field: annotation.path.clone(), + ..Default::default() + }) + .collect(); + enrich_issues_with_catalog(&mut issues); + issues +} + +#[cfg(test)] +mod unknown_annotation_messages { + use super::{pb, unknown_annotation_type_issues}; + use crate::type_libraries::{LibraryConsultation, UnknownAnnotation, UnknownAnnotationTypes}; + + fn found(libraries: LibraryConsultation) -> UnknownAnnotationTypes { + UnknownAnnotationTypes { + annotations: vec![UnknownAnnotation { + type_url: "type.googleapis.com/acme.notes.v1.ReviewNote".into(), + path: "metadata[0]".into(), + }], + libraries, + } + } + + #[test] + fn a_compiler_outage_is_not_described_as_a_compile_failure() { + let subject = pb::EntityRef::default(); + let outage = unknown_annotation_type_issues( + &subject, + &found(LibraryConsultation::CompilerUnavailable), + ); + let broken = + unknown_annotation_type_issues(&subject, &found(LibraryConsultation::DidNotCompile)); + assert!(outage[0].message.contains("type compiler was unavailable")); + assert!(!outage[0].message.contains("do not compile together")); + assert!(broken[0].message.contains("do not compile together")); + } +} + /// Decodes a `metadata` entry as PersonalDataAnnotation. Returns `None` /// when the entry is not a PersonalDataAnnotation at all (not our /// concern), `Some(Err(()))` when the type_url matches but the payload @@ -1188,10 +1250,7 @@ fn walk_personal_data_fields<'a>( fn decode_personal_data_annotation( any: &prost_types::Any, ) -> Option> { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.PersonalDataAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::PERSONAL_DATA.matches(any) { return None; } Some( @@ -3589,9 +3648,7 @@ pub fn validate_model_content( let mut present = false; let mut doc_ok = false; for any in &rm.metadata { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.SharedConsumersAnnotation") + if !crate::annotation_type::BuiltinAnnotation::SHARED_CONSUMERS.matches(any) { continue; } @@ -4214,10 +4271,7 @@ pub fn validate_model_content( } let metadata = entity_metadata(&se.entity); for any in metadata { - if !any - .type_url - .ends_with("trogonatlas.annotation.v1alpha1.CodeRefAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::CODE_REF.matches(any) { continue; } let Ok(cr) = @@ -4281,10 +4335,7 @@ pub fn validate_model_content( // Malformed payloads are treated as absent: they must not panic. let decode_lifecycle_status = |metadata: &[prost_types::Any]| -> Option { for any in metadata { - if !any - .type_url - .ends_with("trogonatlas.annotation.v1alpha1.LifecycleAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::LIFECYCLE.matches(any) { continue; } if let Ok(ann) = @@ -4393,10 +4444,7 @@ pub fn validate_model_content( // findings accidentally. let decode_pinned_ann = |metadata: &[prost_types::Any]| -> Option { for any in metadata { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.PinnedRefAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::PINNED_REF.matches(any) { continue; } if let Ok(ann) = @@ -5478,8 +5526,7 @@ pub fn validate_model_content( }) .filter(|ui| { ui.metadata.iter().any(|m| { - m.type_url - .ends_with("/trogonatlas.eventmodel.v1alpha1.MixedSurfaceAnnotation") + crate::annotation_type::BuiltinAnnotation::MIXED_SURFACE.matches(m) && ::decode(m.value.as_slice()) .is_ok_and(|a| is_stated_reason(&a.doc)) }) @@ -5586,10 +5633,7 @@ pub fn validate_model_content( } let mut names: HashSet = HashSet::new(); for any in &lane.metadata { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.Uuidv5IdentityAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::UUIDV5_IDENTITY.matches(any) { continue; } if let Ok(a) = @@ -6901,10 +6945,7 @@ pub fn validate_model_content( let mut orphan_doc: Option = None; let mut orphan_present = false; for any in metadata { - if !any - .type_url - .ends_with("trogonatlas.eventmodel.v1alpha1.OrphanAnnotation") - { + if !crate::annotation_type::BuiltinAnnotation::ORPHAN.matches(any) { continue; } orphan_present = true; diff --git a/rsworkspace/crates/trogon-atlas-server/tests/unknown_annotation_types.rs b/rsworkspace/crates/trogon-atlas-server/tests/unknown_annotation_types.rs new file mode 100644 index 00000000..75ead4d2 --- /dev/null +++ b/rsworkspace/crates/trogon-atlas-server/tests/unknown_annotation_types.rs @@ -0,0 +1,385 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::sync::Arc; + +use pb::event_model_service_server::EventModelService; +use prost::Message; +use tonic::Request; +use trogon_atlas_proto as pb; +use trogon_atlas_server::{metrics::ServerMetrics, service::EventModelServiceImpl}; + +const BUILT_IN: &str = "type.googleapis.com/trogonatlas.eventmodel.v1alpha1.OrphanAnnotation"; +const TENANT: &str = "type.googleapis.com/acme.notes.v1.ReviewNote"; +const MISSPELLED: &str = "type.googleapis.com/trogonatlas.eventmodel.v1alpha1.OrphanAnotation"; +const TIMESTAMP: &str = "type.googleapis.com/google.protobuf.Timestamp"; +const NOTE: &str = "type.googleapis.com/trogonatlas.annotation.v1alpha1.NoteAnnotation"; + +async fn service() -> EventModelServiceImpl { + let nats = trogon_atlas_testsupport::shared().await; + EventModelServiceImpl::try_new(Arc::new(nats.store().await), ServerMetrics::default()).unwrap() +} + +fn id(namespace: &str, slug: &str) -> pb::Id { + pb::Id { + namespace: namespace.into(), + slug: slug.into(), + version: 1, + } +} + +fn annotation(type_url: &str) -> prost_types::Any { + prost_types::Any { + type_url: type_url.into(), + value: pb::OrphanAnnotation { + doc: "kept for replaying v1 streams".into(), + } + .encode_to_vec(), + } +} + +fn notes_library(namespace: &str) -> pb::Entity { + pb::Entity { + kind: Some(pb::entity::Kind::TypeLibrary(pb::TypeLibrary { + id: Some(id(namespace, "acme.notes")), + title: "Notes".into(), + files: vec![pb::ProtoSourceFile { + path: "acme/notes/v1/notes.proto".into(), + content: "syntax = \"proto3\";\npackage acme.notes.v1;\nmessage ReviewNote { string doc = 1; }\n".into(), + }], + provenance: Some(pb::TypeLibraryProvenance { + source: Some(pb::type_library_provenance::Source::Git( + pb::type_library_provenance::GitSnapshot { + repository_url: "https://example.com/acme/notes.git".into(), + r#ref: "main".into(), + commit: "0123456789abcdef".into(), + subdirectory: "proto".into(), + }, + )), + }), + ..Default::default() + })), + ..Default::default() + } +} + +fn annotated_event(namespace: &str, type_urls: &[&str]) -> pb::Entity { + pb::Entity { + kind: Some(pb::entity::Kind::Event(pb::Event { + id: Some(id(namespace, "placed")), + title: "placed".into(), + metadata: type_urls.iter().map(|url| annotation(url)).collect(), + ..Default::default() + })), + ..Default::default() + } +} + +fn put_request(entity: pb::Entity, validate_only: bool) -> pb::PutEntityRequest { + pb::PutEntityRequest { + entity: Some(entity), + validate_only, + ..Default::default() + } +} + +async fn put( + service: &EventModelServiceImpl, + entity: pb::Entity, + validate_only: bool, +) -> pb::PutEntityResponse { + service + .put_entity(Request::new(put_request(entity, validate_only))) + .await + .unwrap() + .into_inner() +} + +async fn batch_put( + service: &EventModelServiceImpl, + entities: Vec, + validate_only: bool, +) -> Vec { + service + .batch_mutate(Request::new(pb::BatchMutateRequest { + ops: entities + .into_iter() + .map(|entity| pb::BatchMutateOp { + op: Some(pb::batch_mutate_op::Op::Put(put_request(entity, false))), + }) + .collect(), + validate_only, + ..Default::default() + })) + .await + .unwrap() + .into_inner() + .results + .into_iter() + .map(|result| match result.result { + Some(pb::batch_mutate_op_result::Result::Put(put)) => put, + other => panic!("expected a put result, got {other:?}"), + }) + .collect() +} + +/// Each unknown-annotation warning as (severity, subject, path, message names the url). +fn unknown_annotations(issues: &[pb::ValidationIssue]) -> Vec<(i32, pb::Id, String, bool)> { + issues + .iter() + .filter(|i| i.code == "UNKNOWN_ANNOTATION_TYPE") + .map(|i| { + ( + i.severity, + i.subject.as_ref().unwrap().id.clone().unwrap(), + i.subject_field.clone(), + i.message.contains(MISSPELLED), + ) + }) + .collect() +} + +fn misspelled_at(namespace: &str, path: &str) -> Vec<(i32, pb::Id, String, bool)> { + vec![( + pb::validation_issue::Severity::Warning as i32, + id(namespace, "placed"), + path.into(), + true, + )] +} + +#[tokio::test(flavor = "multi_thread")] +async fn put_warns_about_a_well_known_type_used_as_an_annotation() { + let service = service().await; + let ns = "uat-timestamp"; + + let response = put(&service, annotated_event(ns, &[TIMESTAMP]), false).await; + + assert_eq!( + unknown_annotations(&response.validation), + vec![( + pb::validation_issue::Severity::Warning as i32, + id(ns, "placed"), + "metadata[0]".into(), + false, + )] + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn put_accepts_a_built_in_annotation_outside_the_validation_list() { + let service = service().await; + let ns = "uat-note"; + + let response = put(&service, annotated_event(ns, &[NOTE]), false).await; + + assert_eq!(unknown_annotations(&response.validation), Vec::new()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn batch_delete_of_a_library_warns_about_its_annotation() { + let service = service().await; + let ns = "uat-deleted-library"; + put(&service, notes_library(ns), false).await; + + let response = service + .batch_mutate(Request::new(pb::BatchMutateRequest { + ops: vec![ + pb::BatchMutateOp { + op: Some(pb::batch_mutate_op::Op::Delete(pb::DeleteEntityRequest { + kind: pb::EntityKind::TypeLibrary as i32, + id: Some(id(ns, "acme.notes")), + ..Default::default() + })), + }, + pb::BatchMutateOp { + op: Some(pb::batch_mutate_op::Op::Put(put_request( + annotated_event(ns, &[TENANT]), + false, + ))), + }, + ], + ..Default::default() + })) + .await + .unwrap() + .into_inner(); + let put = match &response.results[1].result { + Some(pb::batch_mutate_op_result::Result::Put(put)) => put, + other => panic!("expected a put result, got {other:?}"), + }; + + assert_eq!( + unknown_annotations(&put.validation), + vec![( + pb::validation_issue::Severity::Warning as i32, + id(ns, "placed"), + "metadata[0]".into(), + false, + )] + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn put_warns_only_about_the_misspelled_annotation() { + let service = service().await; + for validate_only in [true, false] { + let ns = format!("uat-put-{validate_only}"); + put(&service, notes_library(&ns), false).await; + + let response = put( + &service, + annotated_event(&ns, &[BUILT_IN, TENANT, MISSPELLED]), + validate_only, + ) + .await; + + assert_eq!( + unknown_annotations(&response.validation), + misspelled_at(&ns, "metadata[2]"), + "validate_only={validate_only}" + ); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn put_warns_about_a_tenant_type_its_namespace_has_no_library_for() { + let service = service().await; + let ns = "uat-no-library"; + + let response = put(&service, annotated_event(ns, &[BUILT_IN, TENANT]), false).await; + + let found: Vec = response + .validation + .iter() + .filter(|i| i.code == "UNKNOWN_ANNOTATION_TYPE") + .map(|i| i.subject_field.clone()) + .collect(); + assert_eq!(found, vec!["metadata[1]".to_owned()]); +} + +#[tokio::test(flavor = "multi_thread")] +async fn batch_mutate_resolves_a_library_written_in_the_same_batch() { + let service = service().await; + for validate_only in [true, false] { + let ns = format!("uat-batch-{validate_only}"); + + let results = batch_put( + &service, + vec![ + notes_library(&ns), + annotated_event(&ns, &[BUILT_IN, TENANT, MISSPELLED]), + ], + validate_only, + ) + .await; + + assert_eq!(unknown_annotations(&results[0].validation), Vec::new()); + assert_eq!( + unknown_annotations(&results[1].validation), + misspelled_at(&ns, "metadata[2]"), + "validate_only={validate_only}" + ); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn batch_mutate_warns_about_misspelled_annotations_nested_in_a_schema() { + let service = service().await; + let ns = "uat-nested"; + let mut event = annotated_event(ns, &[]); + if let Some(pb::entity::Kind::Event(event)) = event.kind.as_mut() { + event.schema = Some(prost_types::Any { + type_url: "type.googleapis.com/trogonatlas.eventmodel.v1alpha1.Schema".into(), + value: pb::Schema { + fields: vec![pb::FieldSpec { + name: "order_id".into(), + r#type: Some(pb::FieldType { + kind: Some(pb::field_type::Kind::String(pb::field_type::StringType {})), + }), + metadata: vec![annotation(MISSPELLED)], + ..Default::default() + }], + ..Default::default() + } + .encode_to_vec(), + }); + } + + let results = batch_put(&service, vec![event], false).await; + + assert_eq!( + unknown_annotations(&results[0].validation), + misspelled_at(ns, "schema.fields[0].metadata[0]") + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn put_checks_annotations_inside_a_tenant_typed_payload() { + let service = service().await; + let ns = "uat-opaque"; + put(&service, envelope_library(ns), false).await; + + let mut event = annotated_event(ns, &[]); + if let Some(pb::entity::Kind::Event(event)) = event.kind.as_mut() { + event.schema = Some(prost_types::Any { + type_url: "type.googleapis.com/acme.notes.v1.Envelope".into(), + value: proto_message(&[(1, proto_any(MISSPELLED))]), + }); + } + + let response = put(&service, event, false).await; + + assert_eq!( + unknown_annotations(&response.validation), + misspelled_at(ns, "schema.metadata[0]") + ); +} + +fn envelope_library(namespace: &str) -> pb::Entity { + pb::Entity { + kind: Some(pb::entity::Kind::TypeLibrary(pb::TypeLibrary { + id: Some(id(namespace, "acme.notes")), + title: "Notes".into(), + files: vec![pb::ProtoSourceFile { + path: "acme/notes/v1/notes.proto".into(), + content: "syntax = \"proto3\";\npackage acme.notes.v1;\nimport \"google/protobuf/any.proto\";\nmessage Envelope { repeated google.protobuf.Any metadata = 1; }\n".into(), + }], + provenance: Some(pb::TypeLibraryProvenance { + source: Some(pb::type_library_provenance::Source::Git( + pb::type_library_provenance::GitSnapshot { + repository_url: "https://example.com/acme/notes.git".into(), + r#ref: "main".into(), + commit: "0123456789abcdef".into(), + subdirectory: "proto".into(), + }, + )), + }), + ..Default::default() + })), + ..Default::default() + } +} + +fn proto_any(type_url: &str) -> Vec { + proto_message(&[ + (1, type_url.as_bytes().to_vec()), + ( + 2, + pb::OrphanAnnotation { + doc: "kept for replaying v1 streams".into(), + } + .encode_to_vec(), + ), + ]) +} + +fn proto_message(fields: &[(u32, Vec)]) -> Vec { + let mut out = Vec::new(); + for (field, value) in fields { + out.push(u8::try_from((field << 3) | 2).unwrap()); + let len = u8::try_from(value.len()).unwrap(); + out.push(len); + out.extend(value); + } + out +}