Fix exponential memory use from independent Fn::If conditions and YAML aliases

src/bindings-jvm/tests/kotlin/src/test/kotlin/SmokeTest.kt

@@ -10,6 +10,7 @@ import software.amazon.cloudformation.validate.engine.*
 import software.amazon.cloudformation.validate.gson.buildBindingsGson
 import software.amazon.cloudformation.validate.rules.*
 import software.amazon.cloudformation.validate.schemavalidator.SchemaValidatorConfig
+import software.amazon.cloudformation.validate.templatemodel.JsonValueEnum
 import java.io.File
 
 class SmokeTest {
@@ -573,6 +574,20 @@ class SmokeTest {
         assertEquals(1L, counts["errors"])
     }
 
+    @Test
+    fun explicitJsonNullContextSurvivesFfi() {
+        val report = CEL.validateTemplate(
+            templateFile("bad/resources/properties/list_duplicates.yaml"),
+            detailedConfig(),
+        )
+        val diagnostic = report.diagnostics.first { it.entity?.logicalId == "IamGroupWithConditions" }
+
+        assertTrue(
+            diagnostic.context?.actualValue is JsonValueEnum.Null,
+            "an explicit JSON null must remain distinct from an absent violation-context value",
+        )
+    }
+
     @TestFactory
     fun regoDetailedMatchesSnapshot(): List<DynamicTest> = snapshotDetailedTests("rego", REGO)
 
@@ -698,10 +713,24 @@ class SmokeTest {
         report["metadata"] = metadata
     }
 
+    @Suppress("UNCHECKED_CAST")
+    private fun stripNullActualValuesOmittedByGson(report: MutableMap<String, Any?>) {
+        val diagnostics = report["diagnostics"] as? List<Map<String, Any?>> ?: return
+        report["diagnostics"] = diagnostics.map { diagnostic ->
+            val normalized = LinkedHashMap(diagnostic)
+            val context = diagnostic["context"] as? Map<String, Any?>
+            if (context?.containsKey("actualValue") == true && context["actualValue"] == null) {
+                normalized["context"] = LinkedHashMap(context).apply { remove("actualValue") }
+            }
+            normalized
+        }
+    }
+
     @Suppress("UNCHECKED_CAST")
     private fun stripSnapshotExcludedFields(report: Map<String, Any?>, filePath: String? = null): Map<String, Any?> {
         val out = LinkedHashMap(report)
         excludeSnapshotRules(out)
+        stripNullActualValuesOmittedByGson(out)
         if (filePath != null) out["filePath"] = filePath
         out.remove("version")
         out.remove("performance")

src/bindings-rust/tests/smoke.rs

@@ -1,8 +1,12 @@
+use cloudformation_validate::template_model::{resolved_value_to_json, resolver::ResolvedValue};
 use cloudformation_validate::{
     AdditionalSchemaSource, CelEngine, CompositeEngine, CompositeEngineConfig, EngineConfig, ExternalRuleSource,
     RegoEngine, ReportStatus, SchemaValidator, SchemaValidatorConfig, SemanticModel, Severity, ValidateConfig,
     ValidationEngine, ValidationReport, validate_bytes_with_path, version,
 };
+use std::collections::HashMap;
+
+type Scenario = (ResolvedValue, HashMap<String, bool>);
 
 const GOOD_TEMPLATE: &[u8] = br#"
 AWSTemplateFormatVersion: "2010-09-09"
@@ -19,6 +23,22 @@ Resources:
       BucketName: my-test-bucket
 "#;
 
+const CONDITIONAL_BUCKET_NAME: &[u8] = br#"
+Parameters:
+  Stage:
+    Type: String
+Conditions:
+  IsProduction: !Equals [!Ref Stage, production]
+Resources:
+  MyBucket:
+    Type: AWS::S3::Bucket
+    Properties:
+      BucketName: !If [IsProduction, production-bucket, staging-bucket]
+"#;
+
+const BUCKET_NAME_PATH: &str = "Properties.BucketName";
+const IS_PRODUCTION: &str = "IsProduction";
+
 const TEMPLATE_WITH_OVERLAY_PROPERTY: &[u8] = br#"
 Resources:
   Function:
@@ -90,6 +110,13 @@ fn diagnostic_signatures(report: &ValidationReport) -> Vec<(String, Severity, St
         .collect()
 }
 
+fn rendered_scenarios(scenarios: &[Scenario]) -> Vec<(String, Option<bool>)> {
+    scenarios
+        .iter()
+        .map(|(value, conditions)| (resolved_value_to_json(value).to_string(), conditions.get(IS_PRODUCTION).copied()))
+        .collect()
+}
+
 fn all_engines(config: EngineConfig) -> Vec<Box<dyn ValidationEngine>> {
     // The composite engine takes its own config type. The built-in, default,
     // good, additional-schema, and Guard matrices that use this helper never
@@ -293,6 +320,50 @@ fn semantic_model_is_available_through_the_facade() {
     assert!(model.transforms.is_empty());
 }
 
+#[test]
+fn semantic_model_resolve_scenarios_returns_owned_branch_scenarios() {
+    let model = SemanticModel::from_bytes(CONDITIONAL_BUCKET_NAME).expect("template must parse");
+
+    let scenarios: Vec<Scenario> = model.resolve_scenarios("MyBucket", BUCKET_NAME_PATH);
+
+    assert_eq!(
+        rendered_scenarios(&scenarios),
+        vec![(r#""production-bucket""#.to_string(), Some(true)), (r#""staging-bucket""#.to_string(), Some(false))],
+        "each Fn::If branch must be one owned scenario paired with the condition value that selects it"
+    );
+}
+
+#[test]
+fn semantic_model_resolve_scenarios_with_limit_returns_owned_scenarios_and_curtailment() {
+    let model = SemanticModel::from_bytes(CONDITIONAL_BUCKET_NAME).expect("template must parse");
+
+    let (scenarios, was_curtailed): (Vec<Scenario>, bool) =
+        model.resolve_scenarios_with_limit("MyBucket", BUCKET_NAME_PATH, 1);
+
+    assert_eq!(
+        rendered_scenarios(&scenarios),
+        vec![(r#""production-bucket""#.to_string(), Some(true))],
+        "a per-value limit of one must keep only the first branch scenario"
+    );
+    assert!(was_curtailed, "omitting the second branch scenario must be reported as curtailment");
+}
+
+#[test]
+fn semantic_model_resolve_properties_scenarios_returns_owned_whole_properties_scenarios() {
+    let model = SemanticModel::from_bytes(CONDITIONAL_BUCKET_NAME).expect("template must parse");
+
+    let scenarios: Vec<Scenario> = model.resolve_properties_scenarios("MyBucket");
+
+    assert_eq!(
+        rendered_scenarios(&scenarios),
+        vec![
+            (r#"{"BucketName":"production-bucket"}"#.to_string(), Some(true)),
+            (r#"{"BucketName":"staging-bucket"}"#.to_string(), Some(false)),
+        ],
+        "the whole Properties object must expand into one owned scenario per Fn::If branch"
+    );
+}
+
 #[test]
 fn composite_default_produces_the_same_builtin_diagnostics_as_the_cel_engine() {
     let schema_validator = SchemaValidator::default();

src/cel-engine/src/rules/best_practices.rs

@@ -1,5 +1,5 @@
 use super::patterns::AMI_ID_RE;
-use super::resources_extra::{scenario_has_effective_property, scenario_is_reachable};
+use super::resources_extra::scenario_is_reachable;
 use super::{EvalContext, NativeRuleRegistry};
 use data_source::rule_data::{PathSegment, ResourcePropertyPath};
 use diagnostics::Diagnostic;
@@ -126,11 +126,16 @@ fn rds_storage_encryption_scenario_warns(properties: &ResolvedValue) -> bool {
 }
 
 fn rds_storage_encryption_warns(model: &SemanticModel, resource_id: &str) -> bool {
-    model.resolve_properties_scenarios(resource_id).into_iter().any(|(properties, conditions)| {
-        let assumptions: Vec<(String, bool)> = conditions.into_iter().collect();
-        (assumptions.is_empty() || model.conditions.is_satisfiable(&assumptions))
-            && rds_storage_encryption_scenario_warns(&properties)
-    })
+    let mut inspected_fields = vec!["Engine", "StorageEncrypted"];
+    inspected_fields.extend(RDS_INHERITED_OR_IGNORED_ENCRYPTION_FIELDS);
+    model.resolve_properties_scenarios_projected(resource_id, &inspected_fields).iter().any(
+        |(properties, conditions)| {
+            let assumptions: Vec<(String, bool)> =
+                conditions.iter().map(|(name, value)| (name.clone(), *value)).collect();
+            (assumptions.is_empty() || model.conditions.is_satisfiable(&assumptions))
+                && rds_storage_encryption_scenario_warns(properties)
+        },
+    )
 }
 
 fn eval_best_practices(ctx: &EvalContext) -> Vec<Diagnostic> {
@@ -519,9 +524,10 @@ None,
                 .get(&res.resource_type)
                 .map(|entry| entry.properties.iter().any(|p| p == "Tags"))
                 .unwrap_or(false);
-            let tags_missing = m.resolve_properties_scenarios(name).iter().any(|(properties, conditions)| {
-                scenario_is_reachable(m, name, conditions) && !scenario_has_effective_property(properties, "Tags")
-            });
+            let tags_missing = m
+                .property_absence_scenarios(name, "Tags")
+                .iter()
+                .any(|conditions| scenario_is_reachable(m, name, conditions));
             if supports_tags && tags_missing {
                 out.push(make_resource_diagnostic(
                     "I9040",