fix(header-chain): preserve auxiliary capacity and diversity

crates/xtask/src/header_conformance.rs

@@ -10,7 +10,7 @@ use std::{
 use serde::Deserialize;
 
 const SUPPORTED_SPEC_VERSION: &str = "1.4";
-const EXPECTED_RULE_COUNT: usize = 116;
+const EXPECTED_RULE_COUNT: usize = 117;
 const SPEC_PATH: &str = "docs/specs/fork-aware-header-chain-engine.md";
 const MANIFEST_PATH: &str = "crates/zakura-header-chain/conformance.toml";
 const MAX_DOCUMENT_BYTES: u64 = 2 * 1024 * 1024;

crates/zakura-header-chain/conformance.toml

@@ -729,6 +729,14 @@ owner = "zakura_state::service::finalized_state::VctAuxiliaryWindow"
 tests = ["IN-06::vct_boundary_failure_attribution_never_weakens_authenticated_evidence"]
 networks = ["mainnet", "testnet", "custom"]
 
+[[rule]]
+id = "LC-AUX-06"
+name = "Semantic auxiliary capacity"
+status = "implemented"
+owner = "zakura_header_chain::HeaderChainEngine::plan_transition"
+tests = ["IN-06::auxiliary_admission_preserves_semantic_diversity_under_duplicate_delivery_pressure", "DG-06::projected_auxiliary_rows_cannot_duplicate_a_semantic_payload", "DG-06::projected_per_header_auxiliary_count_must_stay_within_the_plan_limit"]
+networks = ["mainnet", "testnet", "custom"]
+
 [[rule]]
 id = "LC-WIRE-01"
 name = "Bounded decoding"

crates/zakura-header-chain/src/config.rs

@@ -42,8 +42,8 @@ pub const MAX_NON_FINALIZED_NODES_V1: usize = 65_536;
 pub const MAX_STAGED_TARGETS_V1: usize = 16;
 /// Exact v1 maximum prepared headers admitted by one transition.
 pub const MAX_HEADERS_PER_TRANSITION_V1: usize = 4_000;
-/// Exact v1 maximum auxiliary deliveries retained for one header.
-pub const MAX_AUX_DELIVERIES_PER_HEADER_V1: usize = 16;
+/// Exact v1 maximum distinct auxiliary payloads retained for one header.
+pub const MAX_AUX_DELIVERIES_PER_HEADER_V1: usize = 32;
 /// Exact v1 maximum auxiliary deliveries retained across the graph.
 pub const MAX_AUX_DELIVERIES_TOTAL_V1: usize = MAX_NON_FINALIZED_NODES_V1;
 /// Full-state fork policy plus one independent selected header tip set the candidate-tip cap.
@@ -456,7 +456,7 @@ pub struct EngineLimits {
     pub max_non_finalized_nodes: NonZeroUsize,
     /// Maximum prepared headers accepted before any batch-proportional work.
     pub max_headers_per_transition: NonZeroUsize,
-    /// Maximum fixed-size auxiliary records retained for one header.
+    /// Maximum distinct auxiliary payloads retained for one header.
     pub max_aux_deliveries_per_header: NonZeroUsize,
     /// Maximum fixed-size auxiliary records retained across the graph.
     pub max_aux_deliveries_total: NonZeroUsize,
@@ -497,7 +497,7 @@ const _: () = assert!(MAX_CANDIDATE_TIPS_V1 == 11);
 const _: () = assert!(MAX_NON_FINALIZED_NODES_V1 == 65_536);
 const _: () = assert!(MAX_STAGED_TARGETS_V1 == 16);
 const _: () = assert!(MAX_HEADERS_PER_TRANSITION_V1 == 4_000);
-const _: () = assert!(MAX_AUX_DELIVERIES_PER_HEADER_V1 == 16);
+const _: () = assert!(MAX_AUX_DELIVERIES_PER_HEADER_V1 == 32);
 const _: () = assert!(MAX_AUX_DELIVERIES_TOTAL_V1 == 65_536);
 const _: () = assert!(MAX_RETENTION_REFERENCES_V1 == 27);
 
@@ -519,6 +519,8 @@ mod tests {
         assert_eq!(limits.local_finality_depth.get(), 1_000);
         assert_eq!(limits.max_candidate_tips.get(), 11);
         assert_eq!(limits.max_non_finalized_nodes.get(), 65_536);
+        assert_eq!(limits.max_aux_deliveries_per_header.get(), 32);
+        assert_eq!(limits.max_aux_deliveries_total.get(), 65_536);
         assert_eq!(
             limits.max_retention_references.get(),
             MAX_STAGED_TARGETS_V1 + limits.max_candidate_tips.get(),

crates/zakura-header-chain/src/transition/invariants/checks/auxiliary.rs

@@ -60,6 +60,17 @@ pub(crate) fn verify_aux<G: HeaderGraphView>(
             AuxDelta::Delete { .. } => None,
         })
         .collect();
+    let projected_aux_count = engine_before_commit
+        .aux_delivery_count()
+        .saturating_sub(deleted_ids.len())
+        .saturating_add(
+            puts.keys()
+                .filter(|delivery_id| engine_before_commit.aux_delivery(**delivery_id).is_none())
+                .count(),
+        );
+    if projected_aux_count > plan.limits.max_aux_deliveries_total.get() {
+        return Err(InvariantViolation::Limits);
+    }
     let mut nodes: Vec<&HeaderNode> = match mode {
         #[cfg(any(test, feature = "fuzz-impl"))]
         VerificationMode::Exhaustive => graph.view_header_nodes(),
@@ -88,6 +99,9 @@ pub(crate) fn verify_aux<G: HeaderGraphView>(
     };
     nodes.sort_unstable_by_key(|node| node.hash.0);
     for node in nodes {
+        if node.aux_delivery_ids.len() > plan.limits.max_aux_deliveries_per_header.get() {
+            return Err(InvariantViolation::Limits);
+        }
         let mut deliveries = engine_before_commit.aux_deliveries(node.hash).to_vec();
         deliveries.retain(|delivery| !deleted_ids.contains(&delivery.delivery_id));
         for delivery in puts
@@ -104,6 +118,22 @@ pub(crate) fn verify_aux<G: HeaderGraphView>(
                 return Err(InvariantViolation::Auxiliary(node.hash));
             }
         }
+        for delivery in puts.values().filter(|delivery| {
+            delivery.header_hash == node.hash
+                && engine_before_commit
+                    .aux_delivery(delivery.delivery_id)
+                    .is_none()
+        }) {
+            if deliveries.iter().any(|other| {
+                other.delivery_id != delivery.delivery_id
+                    && (other.semantic_fingerprint() == delivery.semantic_fingerprint()
+                        || (delivery.tree_aux.is_some()
+                            && other.tree_aux.is_some()
+                            && other.source == delivery.source))
+            }) {
+                return Err(InvariantViolation::Auxiliary(node.hash));
+            }
+        }
         if node.aux_delivery_ids.iter().any(|delivery_id| {
             !deliveries
                 .iter()
@@ -132,6 +162,8 @@ pub(crate) fn verify_aux<G: HeaderGraphView>(
 
 #[cfg(test)]
 mod tests {
+    use std::num::NonZeroUsize;
+
     use super::super::super::test_support::{
         candidate_with_delta, delivery, fixture, no_change_candidate, projected_graph,
     };
@@ -190,4 +222,139 @@ mod tests {
             ]
         );
     }
+
+    #[test]
+    fn projected_auxiliary_total_must_stay_within_the_plan_limit() {
+        let fixture = fixture(EngineMode::HeadersOnly);
+        let ids = [
+            EvidenceId::from_digest([0x73; 32]),
+            EvidenceId::from_digest([0x74; 32]),
+        ];
+        let mut overlay = GraphOverlay::new(fixture.engine.graph());
+        for id in ids {
+            overlay
+                .record_auxiliary_evidence_delivery(fixture.child.hash, id)
+                .expect("the fixture child accepts the delivery identity");
+        }
+        let mut plan = candidate_with_delta(&fixture.engine, overlay.delta());
+        plan.change_set.aux_changes = ids
+            .into_iter()
+            .map(|id| AuxDelta::Put(Box::new(delivery(&fixture.engine, fixture.child.hash, id))))
+            .collect();
+        plan.limits.max_aux_deliveries_per_header = NonZeroUsize::new(2).expect("two is nonzero");
+        plan.limits.max_aux_deliveries_total = NonZeroUsize::new(1).expect("one is nonzero");
+
+        assert_eq!(
+            verify_in_both_modes(&fixture, &plan),
+            [
+                Err(InvariantViolation::Limits),
+                Err(InvariantViolation::Limits),
+            ]
+        );
+    }
+
+    #[test]
+    fn projected_per_header_auxiliary_count_must_stay_within_the_plan_limit() {
+        let fixture = fixture(EngineMode::HeadersOnly);
+        let ids = [
+            EvidenceId::from_digest([0x75; 32]),
+            EvidenceId::from_digest([0x76; 32]),
+        ];
+        let mut overlay = GraphOverlay::new(fixture.engine.graph());
+        for id in ids {
+            overlay
+                .record_auxiliary_evidence_delivery(fixture.child.hash, id)
+                .expect("the fixture child accepts the delivery identity");
+        }
+        let mut plan = candidate_with_delta(&fixture.engine, overlay.delta());
+        plan.change_set.aux_changes = ids
+            .into_iter()
+            .map(|id| AuxDelta::Put(Box::new(delivery(&fixture.engine, fixture.child.hash, id))))
+            .collect();
+        plan.limits.max_aux_deliveries_per_header = NonZeroUsize::new(1).expect("one is nonzero");
+        plan.limits.max_aux_deliveries_total = NonZeroUsize::new(2).expect("two is nonzero");
+
+        assert_eq!(
+            verify_in_both_modes(&fixture, &plan),
+            [
+                Err(InvariantViolation::Limits),
+                Err(InvariantViolation::Limits),
+            ]
+        );
+    }
+
+    #[test]
+    fn projected_auxiliary_rows_cannot_duplicate_a_semantic_payload() {
+        let fixture = fixture(EngineMode::HeadersOnly);
+        let ids = [
+            EvidenceId::from_digest([0x77; 32]),
+            EvidenceId::from_digest([0x78; 32]),
+        ];
+        let mut overlay = GraphOverlay::new(fixture.engine.graph());
+        for id in ids {
+            overlay
+                .record_auxiliary_evidence_delivery(fixture.child.hash, id)
+                .expect("the fixture child accepts the delivery identity");
+        }
+        let mut plan = candidate_with_delta(&fixture.engine, overlay.delta());
+        plan.change_set.aux_changes = ids
+            .into_iter()
+            .map(|id| AuxDelta::Put(Box::new(delivery(&fixture.engine, fixture.child.hash, id))))
+            .collect();
+        plan.limits.max_aux_deliveries_per_header = NonZeroUsize::new(2).expect("two is nonzero");
+        plan.limits.max_aux_deliveries_total = NonZeroUsize::new(2).expect("two is nonzero");
+
+        assert_eq!(
+            verify_in_both_modes(&fixture, &plan),
+            [
+                Err(InvariantViolation::Auxiliary(fixture.child.hash)),
+                Err(InvariantViolation::Auxiliary(fixture.child.hash)),
+            ]
+        );
+    }
+
+    #[test]
+    fn projected_auxiliary_rows_allow_one_rooted_payload_per_supplier() {
+        let fixture = fixture(EngineMode::HeadersOnly);
+        let ids = [
+            EvidenceId::from_digest([0x79; 32]),
+            EvidenceId::from_digest([0x7a; 32]),
+        ];
+        let mut overlay = GraphOverlay::new(fixture.engine.graph());
+        for id in ids {
+            overlay
+                .record_auxiliary_evidence_delivery(fixture.child.hash, id)
+                .expect("the fixture child accepts the delivery identity");
+        }
+        let mut deliveries = ids.map(|id| delivery(&fixture.engine, fixture.child.hash, id));
+        for (index, delivery) in deliveries.iter_mut().enumerate() {
+            delivery.tree_aux = Some(crate::TreeAuxRecordV1 {
+                height: fixture.child.height,
+                sapling_root: zakura_chain::sapling::tree::Root::default(),
+                orchard_root: zakura_chain::orchard::tree::Root::default(),
+                ironwood_root: zakura_chain::ironwood::tree::Root::default(),
+                sapling_tx_count: 1,
+                orchard_tx_count: 2,
+                ironwood_tx_count: 3,
+                auth_data_root: zakura_chain::block::merkle::AuthDataRoot::from(
+                    [u8::try_from(index).expect("the fixture index fits in u8"); 32],
+                ),
+            });
+        }
+        let mut plan = candidate_with_delta(&fixture.engine, overlay.delta());
+        plan.change_set.aux_changes = deliveries
+            .into_iter()
+            .map(|delivery| AuxDelta::Put(Box::new(delivery)))
+            .collect();
+        plan.limits.max_aux_deliveries_per_header = NonZeroUsize::new(2).expect("two is nonzero");
+        plan.limits.max_aux_deliveries_total = NonZeroUsize::new(2).expect("two is nonzero");
+
+        assert_eq!(
+            verify_in_both_modes(&fixture, &plan),
+            [
+                Err(InvariantViolation::Auxiliary(fixture.child.hash)),
+                Err(InvariantViolation::Auxiliary(fixture.child.hash)),
+            ]
+        );
+    }
 }

crates/zakura-header-chain/src/transition/planner/admission.rs

@@ -7,7 +7,7 @@ use zakura_chain::block;
 use super::{InvalidTransitionEvidence, LimitViolation, TransitionFailure};
 use crate::{
     BodyWorkOwner, EngineLimits, EngineMetadata, EngineMode, EngineSnapshot, EventAdmission,
-    EvidenceId, FinalityRecord, Frontier, HeaderChainEngine, HeaderSyncWorkOwner, MemHeaderStore,
+    FinalityRecord, Frontier, HeaderChainEngine, HeaderSyncWorkOwner, MemHeaderStore,
     TargetCompletion, TransitionContext, TransitionEvent, TransitionInput,
 };
 
@@ -73,7 +73,7 @@ pub(super) fn authenticate_and_admit(
     }
     validate_retention_references(input, context)?;
     let event = input.event();
-    validate_event_resource_bounds(engine, &event, context.config.limits)?;
+    validate_event_resource_bounds(&event, context.config.limits)?;
     validate_authority(&event, context)?;
     let full_state_authorization_version = context
         .full_state_authority
@@ -138,7 +138,6 @@ pub(super) fn validate_snapshot(
 }
 
 fn validate_event_resource_bounds(
-    engine: &HeaderChainEngine,
     event: &TransitionEvent,
     limits: EngineLimits,
 ) -> Result<(), TransitionFailure> {
@@ -157,27 +156,15 @@ fn validate_event_resource_bounds(
     if insert.aux.len() > limits.max_aux_deliveries_total.get() {
         return Err(TransitionFailure::AuxiliaryLimitExceeded);
     }
-    let mut additions = HashMap::<block::Hash, HashSet<EvidenceId>>::new();
+    let mut deliveries_by_header = HashMap::<block::Hash, HashSet<crate::EvidenceId>>::new();
     for delivery in &insert.aux {
-        additions
+        let delivery_ids = deliveries_by_header
             .entry(delivery.header_hash)
-            .or_default()
-            .insert(delivery.delivery_id);
-    }
-    let mut new_total = engine.aux_delivery_count();
-    for (hash, delivery_ids) in additions {
-        let existing = engine.aux_deliveries(hash);
-        let new_count = delivery_ids
-            .iter()
-            .filter(|id| !existing.iter().any(|row| row.delivery_id == **id))
-            .count();
-        if existing.len().saturating_add(new_count) > limits.max_aux_deliveries_per_header.get() {
+            .or_default();
+        delivery_ids.insert(delivery.delivery_id);
+        if delivery_ids.len() > limits.max_aux_deliveries_per_header.get() {
             return Err(TransitionFailure::AuxiliaryLimitExceeded);
         }
-        new_total = new_total.saturating_add(new_count);
-    }
-    if new_total > limits.max_aux_deliveries_total.get() {
-        return Err(TransitionFailure::AuxiliaryLimitExceeded);
     }
     Ok(())
 }

crates/zakura-header-chain/src/transition/planner/event_effects/header_admission.rs

@@ -210,6 +210,8 @@ pub(super) fn admit_prepared_headers(
         .map(|header| (header.hash, header.height))
         .collect();
     let mut delivery_ids = HashSet::new();
+    let mut admitted_semantic_payloads = HashSet::new();
+    let mut admitted_root_sources = HashSet::new();
     for delivery in &event.aux {
         let expected_height = batch_headers.get(&delivery.header_hash).copied();
         if !delivery_ids.insert(delivery.delivery_id)
@@ -245,7 +247,29 @@ pub(super) fn admit_prepared_headers(
                 .into());
             }
         }
+        let semantic_fingerprint = delivery.semantic_fingerprint();
+        let semantic_payload_exists = admitted_semantic_payloads
+            .contains(&(delivery.header_hash, semantic_fingerprint))
+            || engine
+                .aux_deliveries(delivery.header_hash)
+                .iter()
+                .any(|existing| existing.semantic_fingerprint() == semantic_fingerprint);
+        let rooted_source_exists = delivery.tree_aux.is_some()
+            && (admitted_root_sources.contains(&(delivery.header_hash, delivery.source))
+                || engine
+                    .aux_deliveries(delivery.header_hash)
+                    .iter()
+                    .any(|existing| {
+                        existing.tree_aux.is_some() && existing.source == delivery.source
+                    }));
+        if semantic_payload_exists || rooted_source_exists {
+            continue;
+        }
         projected.record_aux_delivery(*delivery)?;
+        admitted_semantic_payloads.insert((delivery.header_hash, semantic_fingerprint));
+        if delivery.tree_aux.is_some() {
+            admitted_root_sources.insert((delivery.header_hash, delivery.source));
+        }
     }
     Ok(())
 }