eth_state_diff/
pending_queue.rs

1//! Universal delta encoding for SSZ lists that behave as FIFO queues.
2//!
3//! This module handles both pure FIFO queues (e.g., consolidations, withdrawals)
4//! and logically-FIFO queues with rare reordering (e.g., deposits).
5//!
6//! It guarantees correctness by strictly validating item boundaries during overlap
7//! detection and ensuring the remaining base queue matches the target prefix exactly.
8
9use crate::types::{ArchivedQueueDiff, QueueDiff};
10
11/// Finds the first occurrence of an SSZ-encoded queue item within a byte slice,
12/// strictly checking only at valid SSZ item boundaries to prevent false positives.
13fn find_chunk_aligned(haystack: &[u8], needle: &[u8], item_ssz_size: usize) -> Option<usize> {
14    if needle.len() != item_ssz_size {
15        return None;
16    }
17    haystack
18        .chunks_exact(item_ssz_size)
19        .position(|chunk| chunk == needle)
20        .map(|idx| idx * item_ssz_size)
21}
22
23/// Computes the delta between two serialized queue lists.
24///
25/// # Algorithm
26///
27/// 1. Locates the first item of the target queue within the base queue **only at
28///    valid item boundaries**.
29/// 2. **Strictly validates** that the remaining items in the base queue are
30///    exactly identical to the prefix of the target queue.
31/// 3. If valid, emits `QueueDiff::Fifo`. If invalid, emits `QueueDiff::FullReplacement`.
32pub fn diff_queue(base_ssz: &[u8], target_ssz: &[u8], item_ssz_size: usize) -> QueueDiff {
33    // Edge case: target is empty, everything was consumed.
34    if target_ssz.is_empty() {
35        return QueueDiff::Fifo {
36            consumed_count: base_ssz.len() as u32 / item_ssz_size as u32,
37            appended_items: Vec::new(),
38        };
39    }
40
41    // Edge case: base is empty, everything is an append.
42    if base_ssz.is_empty() {
43        return QueueDiff::Fifo {
44            consumed_count: 0,
45            appended_items: target_ssz.to_vec(),
46        };
47    }
48
49    let target_head = &target_ssz[..item_ssz_size];
50
51    match find_chunk_aligned(base_ssz, target_head, item_ssz_size) {
52        Some(byte_offset) => {
53            let remaining_base_bytes = &base_ssz[byte_offset..];
54            let expected_target_prefix_len = remaining_base_bytes.len();
55
56            // CRITICAL VALIDATION: Ensure the rest of the base queue exactly matches
57            // the prefix of the target queue. This catches any reordering edge cases.
58            if expected_target_prefix_len <= target_ssz.len()
59                && &target_ssz[..expected_target_prefix_len] == remaining_base_bytes
60            {
61                let consumed_count = (byte_offset / item_ssz_size) as u32;
62                let appended_items = target_ssz[expected_target_prefix_len..].to_vec();
63
64                QueueDiff::Fifo {
65                    consumed_count,
66                    appended_items,
67                }
68            } else {
69                // Overlap found, but the tails don't match. Fallback to safe replacement.
70                QueueDiff::FullReplacement(target_ssz.to_vec())
71            }
72        }
73        None => {
74            // The first item of the target does not exist anywhere in the base.
75            // This happens if the base was entirely consumed. Fallback to safe replacement.
76            QueueDiff::FullReplacement(target_ssz.to_vec())
77        }
78    }
79}
80
81/// Applies a universal queue delta in place.
82///
83/// # Complexity
84///
85/// - `Fifo`: O(consumed bytes + appended bytes)
86/// - `FullReplacement`: O(target queue size)
87pub fn apply_queue(base: &mut Vec<u8>, delta: &ArchivedQueueDiff, item_ssz_size: usize) {
88    let delta: QueueDiff = rkyv::deserialize::<QueueDiff, rkyv::rancor::Error>(delta)
89        .expect("Failed to deserialize PendingDepositsDiff");
90
91    match delta {
92        QueueDiff::Fifo {
93            consumed_count,
94            appended_items,
95        } => {
96            let bytes_to_drain = consumed_count as usize * item_ssz_size;
97            if bytes_to_drain > base.len() {
98                base.clear();
99            } else {
100                base.drain(..bytes_to_drain);
101            }
102
103            if !appended_items.is_empty() {
104                base.extend_from_slice(&appended_items);
105            }
106        }
107        QueueDiff::FullReplacement(replacement) => {
108            base.clear();
109            if !replacement.is_empty() {
110                base.extend_from_slice(&replacement);
111            }
112        }
113    }
114}
eth_state_diff/pending_queue.rs

eth_state_diff/
pending_queue.rs
