diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index 06a657bf..c9aa3a0b 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -524,18 +524,7 @@ fn delete_only_sources( let mut other: HashSet = HashSet::new(); let mut slices = 0usize; let mut deletes = 0usize; - for entity in by_key.values() { - let Some(pb::entity::Kind::ReadModelSlice(rs)) = entity.kind.as_ref() else { - continue; - }; - let targets = rs - .read_model - .as_ref() - .and_then(|re| re.read_model.as_ref()?.id.as_ref()) - == Some(read_model_id); - if !targets { - continue; - } + for rs in slices_feeding(read_model_id, by_key) { slices += 1; let is_delete = rs.projection_role == pb::ProjectionRole::Delete as i32; if is_delete { @@ -553,6 +542,80 @@ fn delete_only_sources( (deleting, slices > 0 && slices == deletes) } +fn slices_feeding<'a>( + read_model_id: &'a pb::Id, + by_key: &'a HashMap, +) -> impl Iterator { + by_key.values().filter_map(move |entity| { + let Some(pb::entity::Kind::ReadModelSlice(rs)) = entity.kind.as_ref() else { + return None; + }; + let targets = rs + .read_model + .as_ref() + .and_then(|re| re.read_model.as_ref()?.id.as_ref()) + == Some(read_model_id); + targets.then_some(rs) + }) +} + +/// Lowercased names of the fields that the other upsert slices feeding the +/// same read model carry in their source events. +fn sibling_supplied_fields( + slice: &pb::ReadModelSlice, + read_model_id: &pb::Id, + by_key: &HashMap, +) -> HashSet { + // The store keeps every version, so only the latest version of each other + // slice counts; an older one would hide a field that a bump dropped. + let mut latest: HashMap<(&str, &str), u64> = HashMap::new(); + for entity in by_key.values() { + if let Some(pb::entity::Kind::ReadModelSlice(rs)) = entity.kind.as_ref() { + if let Some(id) = rs.id.as_ref() { + let version = latest + .entry((&id.namespace, &id.slug)) + .or_insert(id.version); + *version = (*version).max(id.version); + } + } + } + let own = slice + .id + .as_ref() + .map(|id| (id.namespace.as_str(), id.slug.as_str())); + let target = ( + read_model_id.namespace.as_str(), + read_model_id.slug.as_str(), + ); + by_key + .values() + .filter_map(|entity| match entity.kind.as_ref() { + Some(pb::entity::Kind::ReadModelSlice(rs)) => Some(rs), + _ => None, + }) + .filter(|rs| { + // The target matches by name so a read-model bump keeps siblings, while a + // slice whose latest version feeds another read model stops counting. + let feeds = rs + .read_model + .as_ref() + .and_then(|re| re.read_model.as_ref()?.id.as_ref()) + .is_some_and(|id| (id.namespace.as_str(), id.slug.as_str()) == target); + feeds + && rs.id.as_ref().is_some_and(|id| { + let name = (id.namespace.as_str(), id.slug.as_str()); + Some(name) != own && latest.get(&name) == Some(&id.version) + }) + && rs.projection_role != pb::ProjectionRole::Delete as i32 + }) + .flat_map(|rs| rs.source_events.iter()) + .filter_map(|ee| ee.event.as_ref()?.id.as_ref()) + .filter_map(|id| endpoint_for(pb::EntityKind::Event, id, by_key)) + .flat_map(|endpoint| endpoint.fields) + .map(|f| f.name.to_ascii_lowercase()) + .collect() +} + fn resolve_slice_ref( kind: pb::EntityKind, id: &pb::Id, @@ -667,6 +730,24 @@ fn slice_to_flow( endpoint.fields.retain(|f| key.contains(&f.name)); } } + } else if let Some(read_model_id) = rs + .read_model + .as_ref() + .and_then(|re| re.read_model.as_ref()?.id.as_ref()) + { + // Several upsert slices can each write part of one read model, so a + // field this slice lacks is sourced when a sibling slice carries it. + let supplied = sibling_supplied_fields(rs, read_model_id, by_key); + // An endpoint that siblings cover completely leaves the link, or it + // would read as a read model that declares no fields. + downstream.retain_mut(|endpoint| { + let declared = !endpoint.fields.is_empty(); + endpoint.fields.retain(|f| { + !candidate_sources(&f.name, &upstream).is_empty() + || !supplied.contains(&f.name.to_ascii_lowercase()) + }); + !declared || !endpoint.fields.is_empty() + }); } FlowLink { upstream, @@ -1378,6 +1459,272 @@ mod tests { assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); } + fn sibling_scenario(extra_fields: Vec) -> Vec { + let mut fields = vec![ + field("order_id", string_kind()), + field("total_cents", int_kind()), + field("shipped_at", timestamp_kind()), + ]; + fields.extend(extra_fields); + vec![ + stored(swimlane("order", Vec::new())), + stored(event( + "order-placed", + "order", + vec![ + field("order_id", string_kind()), + field("total_cents", int_kind()), + ], + )), + stored(event( + "order-shipped", + "order", + vec![ + field("order_id", string_kind()), + field("shipped_at", timestamp_kind()), + ], + )), + stored(keyed( + read_model( + "order-summary", + fields, + vec!["order-placed", "order-shipped"], + ), + &["order_id"], + )), + stored(read_model_slice( + "summary-on-placed", + vec!["order-placed"], + "order-summary", + )), + stored(read_model_slice( + "summary-on-shipped", + vec!["order-shipped"], + "order-summary", + )), + stored(event_model( + "ordering", + vec![ + (pb::EntityKind::ReadModelSlice, "summary-on-placed"), + (pb::EntityKind::ReadModelSlice, "summary-on-shipped"), + ], + )), + ] + } + + #[test] + fn upsert_slice_does_not_require_fields_a_sibling_slice_supplies() { + let all = sibling_scenario(Vec::new()); + let (gaps, score) = check_completeness(&event_model_scope("ordering"), &all); + assert!(gaps.is_empty(), "{gaps:?}"); + assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); + } + + #[test] + fn field_no_slice_supplies_is_still_missing_with_siblings() { + let all = sibling_scenario(vec![field("carrier", string_kind())]); + let (gaps, _) = check_completeness(&event_model_scope("ordering"), &all); + let missing: Vec<&str> = gaps + .iter() + .filter(|g| g.kind == pb::completeness_gap::Kind::MissingSource as i32) + .map(|g| g.field_path.as_str()) + .collect(); + assert_eq!(missing, vec!["carrier", "carrier"], "{gaps:?}"); + } + + #[test] + fn older_slice_versions_do_not_supply_fields() { + let mut all = sibling_scenario(Vec::new()); + all.push(stored(event( + "order-dispatched", + "order", + vec![field("order_id", string_kind())], + ))); + let mut v2 = read_model_slice( + "summary-on-shipped", + vec!["order-dispatched"], + "order-summary", + ); + if let Some(pb::entity::Kind::ReadModelSlice(rs)) = v2.kind.as_mut() { + rs.id.as_mut().unwrap().version = 2; + } + all.push(stored(v2)); + all.retain(|se| !matches!(se.entity.kind, Some(pb::entity::Kind::EventModel(_)))); + let mut model = event_model( + "ordering", + vec![ + (pb::EntityKind::ReadModelSlice, "summary-on-placed"), + (pb::EntityKind::ReadModelSlice, "summary-on-shipped"), + ], + ); + if let Some(pb::entity::Kind::EventModel(em)) = model.kind.as_mut() { + em.members[1].id.as_mut().unwrap().version = 2; + } + all.push(stored(model)); + let (gaps, _) = check_completeness(&event_model_scope("ordering"), &all); + let missing: Vec<&str> = gaps + .iter() + .filter(|g| g.kind == pb::completeness_gap::Kind::MissingSource as i32) + .map(|g| g.field_path.as_str()) + .collect(); + assert_eq!(missing, vec!["shipped_at", "shipped_at"], "{gaps:?}"); + } + + fn slice_v2(slug: &str, source: &str, read_model: &str, read_model_version: u64) -> pb::Entity { + let mut entity = read_model_slice(slug, vec![source], read_model); + if let Some(pb::entity::Kind::ReadModelSlice(rs)) = entity.kind.as_mut() { + rs.id.as_mut().unwrap().version = 2; + let target = rs.read_model.as_mut().unwrap().read_model.as_mut().unwrap(); + target.id.as_mut().unwrap().version = read_model_version; + } + entity + } + + fn with_shipped_v2(mut all: Vec, slice: pb::Entity) -> Vec { + all.push(stored(slice)); + all.retain(|se| !matches!(se.entity.kind, Some(pb::entity::Kind::EventModel(_)))); + let mut model = event_model( + "ordering", + vec![ + (pb::EntityKind::ReadModelSlice, "summary-on-placed"), + (pb::EntityKind::ReadModelSlice, "summary-on-shipped"), + ], + ); + if let Some(pb::entity::Kind::EventModel(em)) = model.kind.as_mut() { + em.members[1].id.as_mut().unwrap().version = 2; + } + all.push(stored(model)); + all + } + + #[test] + fn read_model_version_bump_keeps_siblings() { + let mut all = sibling_scenario(Vec::new()); + let mut summary_v2 = keyed( + read_model( + "order-summary", + vec![ + field("order_id", string_kind()), + field("total_cents", int_kind()), + field("shipped_at", timestamp_kind()), + ], + vec!["order-placed", "order-shipped"], + ), + &["order_id"], + ); + if let Some(pb::entity::Kind::ReadModel(rm)) = summary_v2.kind.as_mut() { + rm.id.as_mut().unwrap().version = 2; + } + all.push(stored(summary_v2)); + let all = with_shipped_v2( + all, + slice_v2("summary-on-shipped", "order-shipped", "order-summary", 2), + ); + let (gaps, _) = check_completeness(&event_model_scope("ordering"), &all); + assert_eq!( + count(&gaps, pb::completeness_gap::Kind::MissingSource), + 0, + "{gaps:?}" + ); + } + + #[test] + fn slice_moved_to_another_read_model_stops_supplying() { + let mut all = sibling_scenario(Vec::new()); + all.push(stored(keyed( + read_model( + "shipment-status", + vec![ + field("order_id", string_kind()), + field("shipped_at", timestamp_kind()), + ], + vec!["order-shipped"], + ), + &["order_id"], + ))); + let all = with_shipped_v2( + all, + slice_v2("summary-on-shipped", "order-shipped", "shipment-status", 1), + ); + let (gaps, _) = check_completeness(&event_model_scope("ordering"), &all); + let missing: Vec<&str> = gaps + .iter() + .filter(|g| g.kind == pb::completeness_gap::Kind::MissingSource as i32) + .map(|g| g.field_path.as_str()) + .collect(); + assert_eq!(missing, vec!["shipped_at"], "{gaps:?}"); + } + + #[test] + fn slice_fully_covered_by_siblings_is_not_undeclared() { + let mut all = sibling_scenario(Vec::new()); + all.push(stored(event( + "order-viewed", + "order", + vec![field("viewer", string_kind())], + ))); + all.push(stored(read_model_slice( + "summary-on-viewed", + vec!["order-viewed"], + "order-summary", + ))); + let (gaps, _) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "summary-on-viewed"), + &all, + ); + assert_eq!( + count(&gaps, pb::completeness_gap::Kind::UndeclaredFields), + 0, + "{gaps:?}" + ); + } + + #[test] + fn read_model_without_fields_is_still_undeclared() { + let all = vec![ + stored(swimlane("order", Vec::new())), + stored(event( + "order-placed", + "order", + vec![field("order_id", string_kind())], + )), + stored(read_model("orders", Vec::new(), vec!["order-placed"])), + stored(read_model_slice( + "orders-slice", + vec!["order-placed"], + "orders", + )), + ]; + let (gaps, _) = + check_completeness(&scope(pb::EntityKind::ReadModelSlice, "orders-slice"), &all); + assert_eq!( + count(&gaps, pb::completeness_gap::Kind::UndeclaredFields), + 1, + "{gaps:?}" + ); + } + + #[test] + fn delete_sibling_does_not_supply_fields() { + let mut all = sibling_scenario(Vec::new()); + all.retain(|se| { + !matches!( + se.entity.kind.as_ref(), + Some(pb::entity::Kind::ReadModelSlice(rs)) + if rs.id.as_ref().is_some_and(|i| i.slug == "summary-on-shipped") + ) + }); + all.push(stored(with_role( + read_model_slice("summary-on-shipped", vec!["order-shipped"], "order-summary"), + pb::ProjectionRole::Delete, + ))); + let (gaps, _) = check_completeness(&event_model_scope("ordering"), &all); + assert!( + gaps.iter().any(|g| g.field_path == "shipped_at"), + "{gaps:?}" + ); + } + #[test] fn ui_slice_passes_read_model_fields_through() { let mut all = vec![stored(read_model(