From 4b44221a652f2fa344a9948e7c2e4dd50f0b65c9 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 08:45:46 -0400 Subject: [PATCH 1/5] fix(server): skip non-key read model fields for delete projections Signed-off-by: Yordis Prieto --- .../trogon-atlas-server/src/analysis.rs | 107 +++++++++++++++++- 1 file changed, 106 insertions(+), 1 deletion(-) diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index 8c71a85e..d6f6f0f2 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -587,13 +587,22 @@ fn slice_to_flow( } } } - let downstream: Vec = rs + let mut downstream: Vec = rs .read_model .as_ref() .and_then(|re| re.read_model.as_ref()?.id.clone()) .and_then(|id| endpoint_for(pb::EntityKind::ReadModel, &id, by_key)) .into_iter() .collect(); + if rs.projection_role == pb::ProjectionRole::Delete as i32 { + for endpoint in &mut upstream { + endpoint.source_role = FlowSourceRole::Auxiliary; + } + for endpoint in &mut downstream { + let key = read_model_key(&endpoint.entity_ref, by_key); + endpoint.fields.retain(|f| key.contains(&f.name)); + } + } FlowLink { upstream, downstream, @@ -641,6 +650,22 @@ fn slice_to_flow( } } +fn read_model_key( + entity_ref: &pb::EntityRef, + by_key: &HashMap, +) -> Vec { + let Some(id) = entity_ref.id.as_ref() else { + return Vec::new(); + }; + match by_key + .get(&pb::EntityKey::new(pb::EntityKind::ReadModel, id)) + .and_then(|entity| entity.kind.as_ref()) + { + Some(pb::entity::Kind::ReadModel(rm)) => rm.key.clone(), + _ => Vec::new(), + } +} + fn endpoint_for( kind: pb::EntityKind, id: &pb::Id, @@ -985,6 +1010,86 @@ mod tests { assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); } + fn keyed(mut entity: pb::Entity, key: &[&str]) -> pb::Entity { + if let Some(pb::entity::Kind::ReadModel(rm)) = entity.kind.as_mut() { + rm.key = key.iter().map(|k| (*k).to_string()).collect(); + } + entity + } + + fn with_role(mut entity: pb::Entity, role: pb::ProjectionRole) -> pb::Entity { + if let Some(pb::entity::Kind::ReadModelSlice(rs)) = entity.kind.as_mut() { + rs.projection_role = role as i32; + } + entity + } + + #[test] + fn delete_projection_only_requires_key_fields() { + let all = vec![ + stored(swimlane("cart", Vec::new())), + stored(event( + "cart-checked-out", + "cart", + vec![field("cart_id", string_kind())], + )), + stored(keyed( + read_model( + "open-carts", + vec![ + field("cart_id", string_kind()), + field("item_count", int_kind()), + field("total_cents", int_kind()), + ], + vec!["cart-checked-out"], + ), + &["cart_id"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Delete, + )), + ]; + let (gaps, score) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert!(gaps.is_empty(), "{gaps:?}"); + assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); + } + + #[test] + fn upsert_projection_still_requires_every_field() { + let all = vec![ + stored(swimlane("cart", Vec::new())), + stored(event( + "cart-checked-out", + "cart", + vec![field("cart_id", string_kind())], + )), + stored(keyed( + read_model( + "open-carts", + vec![ + field("cart_id", string_kind()), + field("item_count", int_kind()), + ], + vec!["cart-checked-out"], + ), + &["cart_id"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Upsert, + )), + ]; + let (gaps, _) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); + } + #[test] fn event_model_scope_checks_read_model_members() { let all = vec![ From 2fbae84d00feebb49a95347a750f59c7d88d01f1 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 08:47:08 -0400 Subject: [PATCH 2/5] fix(server): treat UI slices as passthrough in information completeness Signed-off-by: Yordis Prieto --- .../trogon-atlas-server/src/analysis.rs | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index d6f6f0f2..59a76963 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -633,12 +633,20 @@ fn slice_to_flow( .iter() .filter_map(|re| re.read_model.as_ref()?.id.clone()) .filter_map(|id| endpoint_for(pb::EntityKind::ReadModel, &id, by_key)) + .map(|endpoint| FlowEndpoint { + source_role: FlowSourceRole::Auxiliary, + ..endpoint + }) .collect(); let downstream: Vec = us .ui .as_ref() .and_then(|ue| ue.ui.as_ref()?.id.clone()) .and_then(|id| endpoint_for(pb::EntityKind::Ui, &id, by_key)) + .map(|endpoint| FlowEndpoint { + source_role: FlowSourceRole::Auxiliary, + ..endpoint + }) .into_iter() .collect(); FlowLink { @@ -1024,6 +1032,35 @@ mod tests { entity } + fn ui_with_slice(slice: &str, read_model: &str, ui: &str) -> Vec { + vec![ + stored(pb::Entity { + system: None, + kind: Some(pb::entity::Kind::Ui(pb::Ui { + id: Some(id(ui)), + ..Default::default() + })), + }), + stored(pb::Entity { + system: None, + kind: Some(pb::entity::Kind::UiSlice(pb::UiSlice { + id: Some(id(slice)), + source_read_models: vec![pb::ReadModelEdge { + read_model: Some(pb::ReadModelRef { + id: Some(id(read_model)), + }), + ..Default::default() + }], + ui: Some(pb::UiEdge { + ui: Some(pb::UiRef { id: Some(id(ui)) }), + ..Default::default() + }), + ..Default::default() + })), + }), + ] + } + #[test] fn delete_projection_only_requires_key_fields() { let all = vec![ @@ -1090,6 +1127,27 @@ mod tests { assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); } + #[test] + fn ui_slice_passes_read_model_fields_through() { + let mut all = vec![stored(read_model( + "order-summary", + vec![ + field("order_id", string_kind()), + field("total_cents", int_kind()), + ], + Vec::new(), + ))]; + all.extend(ui_with_slice( + "show-order-summary", + "order-summary", + "order-page", + )); + let (gaps, score) = + check_completeness(&scope(pb::EntityKind::UiSlice, "show-order-summary"), &all); + assert!(gaps.is_empty(), "{gaps:?}"); + assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); + } + #[test] fn event_model_scope_checks_read_model_members() { let all = vec![ From 600a7428ae95563427e1080cd12b247683769f15 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 10:04:45 -0400 Subject: [PATCH 3/5] fix(server): keep delete projection source events direct for key sourcing Signed-off-by: Yordis Prieto --- .../trogon-atlas-server/src/analysis.rs | 74 ++++++++++++++++++- .../crates/trogon-atlas-server/src/jev.rs | 4 +- 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index 59a76963..b712321b 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -8,6 +8,7 @@ use crate::graph; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum FlowSourceRole { Direct, + Trigger, Auxiliary, } @@ -334,7 +335,9 @@ fn candidate_sources<'a>( for f in &up.fields { if f.name.to_ascii_lowercase() == target { match up.source_role { - FlowSourceRole::Direct => direct.push((up.entity_ref.clone(), f)), + FlowSourceRole::Direct | FlowSourceRole::Trigger => { + direct.push((up.entity_ref.clone(), f)); + } FlowSourceRole::Auxiliary => auxiliary.push((up.entity_ref.clone(), f)), } } @@ -596,7 +599,9 @@ fn slice_to_flow( .collect(); if rs.projection_role == pb::ProjectionRole::Delete as i32 { for endpoint in &mut upstream { - endpoint.source_role = FlowSourceRole::Auxiliary; + if endpoint.source_role == FlowSourceRole::Direct { + endpoint.source_role = FlowSourceRole::Trigger; + } } for endpoint in &mut downstream { let key = read_model_key(&endpoint.entity_ref, by_key); @@ -1095,6 +1100,71 @@ mod tests { assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); } + #[test] + fn delete_projection_key_in_event_and_lane_state_is_not_ambiguous() { + let all = vec![ + stored(swimlane("cart", vec![field("cart_id", string_kind())])), + stored(event( + "cart-checked-out", + "cart", + vec![ + field("cart_id", string_kind()), + field("note", string_kind()), + ], + )), + stored(keyed( + read_model( + "open-carts", + vec![ + field("cart_id", string_kind()), + field("item_count", int_kind()), + ], + vec!["cart-checked-out"], + ), + &["cart_id"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Delete, + )), + ]; + let (gaps, score) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert!(gaps.is_empty(), "{gaps:?}"); + assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); + } + + #[test] + fn delete_projection_missing_key_in_event_is_reported() { + let all = vec![ + stored(swimlane("cart", Vec::new())), + stored(event( + "cart-checked-out", + "cart", + vec![field("note", string_kind())], + )), + stored(keyed( + read_model( + "open-carts", + vec![field("cart_id", string_kind())], + vec!["cart-checked-out"], + ), + &["cart_id"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Delete, + )), + ]; + let (gaps, _) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); + } + #[test] fn upsert_projection_still_requires_every_field() { let all = vec![ diff --git a/rsworkspace/crates/trogon-atlas-server/src/jev.rs b/rsworkspace/crates/trogon-atlas-server/src/jev.rs index 0025daa6..50562621 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/jev.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/jev.rs @@ -349,7 +349,7 @@ fn endpoint_evidence<'a>( title, doc, source_role: match endpoint.source_role { - FlowSourceRole::Direct => "direct", + FlowSourceRole::Direct | FlowSourceRole::Trigger => "direct", FlowSourceRole::Auxiliary => "auxiliary", }, fields: endpoint.fields.iter().map(field_evidence).collect(), @@ -485,7 +485,7 @@ fn prepare(scope: &pb::AnalysisScope, all: &[StoredEntity]) -> Result = candidates .iter() .filter(|(endpoint, candidate)| { - endpoint.source_role == FlowSourceRole::Direct + endpoint.source_role != FlowSourceRole::Auxiliary && candidate.name.eq_ignore_ascii_case(&field.name) }) .collect(); From 2e1d53f1b861e17126d76d2873130fe0bf36eec2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 10:23:17 -0400 Subject: [PATCH 4/5] fix(server): keep every field required for delete projections on keyless read models Signed-off-by: Yordis Prieto --- .../trogon-atlas-server/src/analysis.rs | 37 ++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index b712321b..0fb70ea4 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -605,7 +605,9 @@ fn slice_to_flow( } for endpoint in &mut downstream { let key = read_model_key(&endpoint.entity_ref, by_key); - endpoint.fields.retain(|f| key.contains(&f.name)); + if !key.is_empty() { + endpoint.fields.retain(|f| key.contains(&f.name)); + } } } FlowLink { @@ -1165,6 +1167,39 @@ mod tests { assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); } + #[test] + fn keyless_delete_projection_keeps_the_normal_requirement() { + let all = vec![ + stored(swimlane("cart", Vec::new())), + stored(event( + "cart-checked-out", + "cart", + vec![field("cart_id", string_kind())], + )), + stored(read_model( + "open-carts", + vec![ + field("cart_id", string_kind()), + field("item_count", int_kind()), + ], + vec!["cart-checked-out"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Delete, + )), + ]; + let (gaps, _) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); + assert_eq!( + count(&gaps, pb::completeness_gap::Kind::UndeclaredFields), + 0 + ); + } + #[test] fn upsert_projection_still_requires_every_field() { let all = vec![ From d474bbaf7a1ccfe538e22c356078d91c4359118c Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Oct 2026 10:26:24 -0400 Subject: [PATCH 5/5] fix(server): hold delete-only source events to the read model key at every scope Signed-off-by: Yordis Prieto --- .../trogon-atlas-server/src/analysis.rs | 158 +++++++++++++++++- 1 file changed, 152 insertions(+), 6 deletions(-) diff --git a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs index 0fb70ea4..06a657bf 100644 --- a/rsworkspace/crates/trogon-atlas-server/src/analysis.rs +++ b/rsworkspace/crates/trogon-atlas-server/src/analysis.rs @@ -485,15 +485,74 @@ fn read_model_to_flow( } } } - let downstream = endpoint_for(pb::EntityKind::ReadModel, read_model_id, by_key) - .into_iter() - .collect(); + let mut downstream: Vec = + endpoint_for(pb::EntityKind::ReadModel, read_model_id, by_key) + .into_iter() + .collect(); + let (delete_only_events, all_delete) = delete_only_sources(read_model_id, by_key); + if all_delete { + upstream.retain(|endpoint| endpoint.source_role == FlowSourceRole::Direct); + for endpoint in &mut downstream { + let key = read_model_key(&endpoint.entity_ref, by_key); + if !key.is_empty() { + endpoint.fields.retain(|f| key.contains(&f.name)); + } + } + } + for endpoint in &mut upstream { + let is_delete_only = endpoint.entity_ref.kind == pb::EntityKind::Event as i32 + && endpoint + .entity_ref + .id + .as_ref() + .is_some_and(|id| delete_only_events.contains(&pb::IdKey::new(id))); + if is_delete_only { + endpoint.source_role = FlowSourceRole::Trigger; + } + } Some(FlowLink { upstream, downstream, }) } +fn delete_only_sources( + read_model_id: &pb::Id, + by_key: &HashMap, +) -> (HashSet, bool) { + let mut deleting: HashSet = HashSet::new(); + 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; + } + slices += 1; + let is_delete = rs.projection_role == pb::ProjectionRole::Delete as i32; + if is_delete { + deletes += 1; + } + let bucket = if is_delete { &mut deleting } else { &mut other }; + bucket.extend( + rs.source_events + .iter() + .filter_map(|ee| ee.event.as_ref()?.id.as_ref()) + .map(pb::IdKey::new), + ); + } + deleting.retain(|id| !other.contains(id)); + (deleting, slices > 0 && slices == deletes) +} + fn resolve_slice_ref( kind: pb::EntityKind, id: &pb::Id, @@ -598,10 +657,9 @@ fn slice_to_flow( .into_iter() .collect(); if rs.projection_role == pb::ProjectionRole::Delete as i32 { + upstream.retain(|endpoint| endpoint.source_role == FlowSourceRole::Direct); for endpoint in &mut upstream { - if endpoint.source_role == FlowSourceRole::Direct { - endpoint.source_role = FlowSourceRole::Trigger; - } + endpoint.source_role = FlowSourceRole::Trigger; } for endpoint in &mut downstream { let key = read_model_key(&endpoint.entity_ref, by_key); @@ -1200,6 +1258,94 @@ mod tests { ); } + fn delete_scenario( + lane_state: Vec, + event_fields: Vec, + ) -> Vec { + vec![ + stored(swimlane("cart", lane_state)), + stored(event("cart-checked-out", "cart", event_fields)), + stored(keyed( + read_model( + "open-carts", + vec![ + field("cart_id", string_kind()), + field("item_count", int_kind()), + ], + vec!["cart-checked-out"], + ), + &["cart_id"], + )), + stored(with_role( + read_model_slice("carts-on-checkout", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Delete, + )), + ] + } + + #[test] + fn delete_projection_does_not_source_key_from_lane_state() { + let all = delete_scenario( + vec![field("cart_id", string_kind())], + vec![field("note", string_kind())], + ); + let (gaps, _) = check_completeness( + &scope(pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + &all, + ); + assert_eq!(count(&gaps, pb::completeness_gap::Kind::MissingSource), 1); + assert!(gaps.iter().any(|g| g.field_path == "cart_id"), "{gaps:?}"); + } + + #[test] + fn delete_projection_in_event_model_scope_only_requires_key() { + let mut all = delete_scenario( + Vec::new(), + vec![ + field("cart_id", string_kind()), + field("note", string_kind()), + ], + ); + all.push(stored(event_model( + "carting", + vec![ + (pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + (pb::EntityKind::ReadModel, "open-carts"), + ], + ))); + let (gaps, score) = check_completeness(&event_model_scope("carting"), &all); + assert!(gaps.is_empty(), "{gaps:?}"); + assert!((score - 1.0).abs() < f32::EPSILON, "score = {score}"); + } + + #[test] + fn event_used_by_upsert_and_delete_keeps_full_requirement_in_event_model_scope() { + let mut all = delete_scenario( + Vec::new(), + vec![ + field("cart_id", string_kind()), + field("note", string_kind()), + ], + ); + all.push(stored(with_role( + read_model_slice("carts-upsert", vec!["cart-checked-out"], "open-carts"), + pb::ProjectionRole::Upsert, + ))); + all.push(stored(event_model( + "carting", + vec![ + (pb::EntityKind::ReadModelSlice, "carts-on-checkout"), + (pb::EntityKind::ReadModelSlice, "carts-upsert"), + (pb::EntityKind::ReadModel, "open-carts"), + ], + ))); + let (gaps, _) = check_completeness(&event_model_scope("carting"), &all); + assert!( + gaps.iter().any(|g| g.field_path == "item_count"), + "{gaps:?}" + ); + } + #[test] fn upsert_projection_still_requires_every_field() { let all = vec![