pdk_cache_lib/

fifo_cache.rs

// Copyright 2023 Salesforce, Inc. All rights reserved.
use log;
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;

use super::{error::CacheError, Cache};
use pdk_core::classy::shared_data::concurrent_shared_data::TransactionStatus;

cfg_if::cfg_if! {
    if #[cfg(test)] {
        use crate::mocks::MockConcurrentSharedData as ConcurrentSharedData;
    } else {
        use pdk_core::classy::shared_data::concurrent_shared_data::ConcurrentSharedData;
    }
}

enum OperationType {
    Insert,
    Remove,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct InternalData {
    max_entries_size: usize,
    fifo: VecDeque<String>,
    last_modified_item_key: String,
}

impl InternalData {
    fn new(max_entries_size: usize, fifo: VecDeque<String>, last_modified_item_key: &str) -> Self {
        Self {
            max_entries_size,
            fifo,
            last_modified_item_key: last_modified_item_key.to_string(),
        }
    }
}

/// A FIFO cache, that shares global state along all incoming requests of a policy.
///
/// As the name suggests, a FIFO eviction strategy is used, meaning that when the maximum number
/// of entries is reached, the first added cache entries will be the first to be evicted.
pub struct FifoCache {
    cache_id: String,
    cache_id_internal_data: String,
    concurrent_shared_data: ConcurrentSharedData,
    max_entries_size: usize,
}

impl FifoCache {
    pub(crate) fn new(
        cache_id: String,
        max_entries_size: usize,
        concurrent_shared_data: ConcurrentSharedData,
    ) -> Self {
        Self {
            cache_id_internal_data: format!("{}-internal_data", &cache_id),
            cache_id,
            max_entries_size,
            concurrent_shared_data,
        }
    }

    /// Get the cache's maximum number of entries available.
    pub fn max_entries_size(&self) -> usize {
        self.max_entries_size
    }

    /// Returns cache ID.
    pub fn cache_id(&self) -> &str {
        self.cache_id.as_str()
    }

    /// Get a reference to the fifo shared data cache's cache id internal data.
    fn cache_id_internal_data(&self) -> &str {
        self.cache_id_internal_data.as_ref()
    }

    /// Creates a key to be used by cache to store data, using cache instance ID as prefix.
    fn cache_key(&self, item_key: &str) -> String {
        format!("{}-{}", self.cache_id(), item_key)
    }

    fn new_internal_data(
        &self,
        fifo: VecDeque<String>,
        last_modified_item_key: &str,
    ) -> InternalData {
        InternalData::new(self.max_entries_size(), fifo, last_modified_item_key)
    }

    fn get_stored_internal_data(&self) -> Option<InternalData> {
        self.concurrent_shared_data
            .get::<InternalData>(self.cache_id_internal_data())
    }

    fn update_internal_data(&self, last_in: String, is_update: bool) {
        log::trace!(
            "Updating cache state data for key: {} (updated: {})",
            &last_in,
            is_update
        );

        if is_update {
            self.remove_item_from_internal_data(&last_in);
        }

        // update variable with shared_data info; otherwise init as empty.
        let mut internal_data: InternalData;
        if let Some(stored_internal_data) = self.get_stored_internal_data() {
            internal_data = stored_internal_data;
        } else {
            internal_data =
                self.new_internal_data(VecDeque::with_capacity(self.max_entries_size()), &last_in);
        }

        // insert in queue
        internal_data.fifo.push_back(last_in.clone());
        internal_data.last_modified_item_key = last_in;

        // Save updated internal data
        self.save_internal_data(&internal_data, OperationType::Insert);

        // check eviction
        self.check_eviction(&mut internal_data);
    }

    fn check_eviction(&self, internal_data: &mut InternalData) {
        let mut is_evicted = false;
        while internal_data.fifo.len() > self.max_entries_size() {
            let first_in = internal_data.fifo.pop_front().unwrap();

            self.evict(&first_in);
            internal_data.last_modified_item_key = first_in;
            is_evicted = true;
        }

        if is_evicted {
            self.save_internal_data(internal_data, OperationType::Remove);
            log::trace!("Cache evicted");
        }
    }

    fn evict(&self, key: &str) {
        let _deleted: Option<Vec<u8>> = self.concurrent_shared_data.remove(key);
    }

    fn save_internal_data(&self, new_internal_data: &InternalData, operation_type: OperationType) {
        // declare unique variable since destructuring assignments are unstable
        // see issue #71126 <https://github.com/rust-lang/rust/issues/71126> for more information
        // related issue: https://github.com/rust-lang/rust/issues/90095
        let insert_result: (TransactionStatus, InternalData) = match operation_type {
            OperationType::Insert => self.concurrent_shared_data.insert(
                self.cache_id_internal_data().to_string(),
                new_internal_data.to_owned(),
                handle_cache_state_consistency_when_insert,
            ),
            OperationType::Remove => self.concurrent_shared_data.insert(
                self.cache_id_internal_data().to_string(),
                new_internal_data.to_owned(),
                handle_cache_state_consistency_when_remove_or_evict,
            ),
        };

        if !matches!(insert_result.0, TransactionStatus::Complete) {
            log::error!(
                "Error updating internal cache state {}: {:?}",
                self.cache_id(),
                insert_result.0
            );
        }
    }

    fn remove_item_from_internal_data(&self, deleted_key: &str) {
        if let Some(mut stored_internal_data) = self.get_stored_internal_data() {
            // Remove from queue
            if let Some(found_index) =
                stored_internal_data
                    .fifo
                    .iter()
                    .enumerate()
                    .find_map(|(index, key)| {
                        if key.as_str() == deleted_key {
                            Some(index)
                        } else {
                            None
                        }
                    })
            {
                stored_internal_data.fifo.remove(found_index);
                stored_internal_data.last_modified_item_key = deleted_key.to_string();

                // Store updated value
                self.save_internal_data(&stored_internal_data, OperationType::Remove);
            }
        } else {
            log::trace!("Stored cache state data not found, nothing to clean from cache state");
        }
    }
}

/// In any case, if prev_value was already inserted or not, deleted or not; we decide override existing value with new one.
fn handle_cache_value_consistency(
    _stored_value: Vec<u8>,
    candidate_value: Vec<u8>,
) -> Option<Vec<u8>> {
    Some(candidate_value)
}

/// This method applies for insertion operations, and adds last inserted into stored state.
fn handle_cache_state_consistency_when_insert(
    stored_state: InternalData,
    candidate_state: InternalData,
) -> Option<InternalData> {
    let mut new_fifo: VecDeque<String> = stored_state.fifo;

    if let Some(index) = new_fifo.iter().enumerate().find_map(|(index, key)| {
        if key.as_str() == candidate_state.last_modified_item_key.as_str() {
            Some(index)
        } else {
            None
        }
    }) {
        new_fifo.remove(index);
    }
    new_fifo.push_back(candidate_state.last_modified_item_key.clone());

    Some(InternalData::new(
        stored_state.max_entries_size,
        new_fifo,
        candidate_state.last_modified_item_key.as_str(),
    ))
}

/// This method applies for removing operations, and removes all elements not present in candidate state from stored state.
fn handle_cache_state_consistency_when_remove_or_evict(
    stored_state: InternalData,
    candidate_state: InternalData,
) -> Option<InternalData> {
    let mut new_fifo: VecDeque<String> = stored_state.fifo;
    new_fifo.retain(|key| key.as_str() != candidate_state.last_modified_item_key.as_str());

    Some(InternalData::new(
        stored_state.max_entries_size,
        new_fifo,
        candidate_state.last_modified_item_key.as_str(),
    ))
}

impl Cache for FifoCache {
    fn save(&self, key: &str, value: Vec<u8>) -> Result<(), CacheError> {
        // store data in shared data
        let cache_key = self.cache_key(key);

        // First, verify if key is already stored in cache,
        // in order to see if we need to remove element from FIFO and then update cache item correctly.
        let is_update = self
            .concurrent_shared_data
            .get::<Vec<u8>>(cache_key.as_str())
            .is_some();

        // After cleaning up FIFO, let's udpate cache item.
        let (tx_status, _saved_data) =
            self.concurrent_shared_data
                .insert(cache_key, value, handle_cache_value_consistency);

        match tx_status {
            TransactionStatus::Complete => {
                self.update_internal_data(self.cache_key(key), is_update);
            }
            _ => {
                return Err(CacheError::SaveError {
                    status: format!("{:?}", tx_status),
                    cache_id: String::from(self.cache_id()),
                })
            }
        }
        Ok(())
    }

    fn get(&self, key: &str) -> Option<Vec<u8>> {
        let cache_key = self.cache_key(key);
        self.concurrent_shared_data
            .get::<Vec<u8>>(cache_key.as_str())
    }

    fn delete(&self, key: &str) -> Option<Vec<u8>> {
        log::trace!("Delete item '{}' from cache", key);
        let delete_key = self.cache_key(key);
        let deleted = self.concurrent_shared_data.remove(&delete_key);
        self.remove_item_from_internal_data(&delete_key);
        deleted
    }

    fn purge(&self) {
        if let Some(cache_internal_data) = self.get_stored_internal_data() {
            // Clear cache items
            cache_internal_data.fifo.iter().for_each(|cache_key| {
                let _removed: Option<Vec<u8>> = self.concurrent_shared_data.remove(cache_key);
            });
            // Clear internal data
            let _internal_data: Option<InternalData> = self
                .concurrent_shared_data
                .remove(self.cache_id_internal_data());
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::VecDeque;

    use mockall::{
        predicate::{always, eq},
        Sequence,
    };
    use serde::{Deserialize, Serialize};

    use super::Cache;
    use super::ConcurrentSharedData as MockConcurrentSharedData;
    use super::{
        handle_cache_state_consistency_when_insert,
        handle_cache_state_consistency_when_remove_or_evict, handle_cache_value_consistency,
        FifoCache, InternalData,
    };
    use pdk_core::classy::shared_data::concurrent_shared_data::TransactionStatus;

    #[derive(Debug, Serialize, Deserialize, PartialEq, Copy, Clone)]
    struct SerializableObject {
        property_one: u64,
        property_two: u64,
    }

    #[test]
    fn save_in_cache_first_time() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_already_exists() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let stored_obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_stored_obj = bincode::serialize(&stored_obj).unwrap();

        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(serialized_stored_obj.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let stored_internal_data = InternalData::new(3, VecDeque::new(), "test-cache-key1");

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_shared_data_fails_storing_value() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_err());
    }

    // TODO: W-10712428 Update test when implement rollback for errors when save internal data.
    #[test]
    fn save_in_cache_when_shared_data_fails_storing_internaldata() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let stored_obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_stored_obj = bincode::serialize(&stored_obj).unwrap();

        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(serialized_stored_obj.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let stored_internal_data = InternalData::new(3, VecDeque::new(), "test-cache-key1");

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_first_time_when_cache_has_elements() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(eq(String::from("test-cache-key1")), always(), always())
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::InternalError, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_cache_is_full_needs_eviction() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
            String::from("test-cache-keyC"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyC");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(eq("test-cache-key1".to_string()), always(), always())
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(
                eq(String::from("test-cache-internal_data")),
                always(),
                always(),
            )
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        // eviction value
        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(1)
            .with(eq("test-cache-keyA"))
            .in_sequence(&mut sequence)
            .returning(move |_key_to_evict| None);

        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
            String::from("test-cache-keyC"),
            String::from("test-cache-key1"),
        ]);
        let _stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        // update after eviction
        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .with(
                eq(String::from("test-cache-internal_data")),
                always(),
                always(),
            )
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_ok());
    }

    #[test]
    fn save_in_cache_when_cache_item_has_version_mismatch() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: Vec<u8>, _handle_consistency| {
                (TransactionStatus::Rejected, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        let value = serialized_obj.as_slice().to_vec();
        assert!(cache.save("key1", value).is_err());
    }

    #[test]
    fn test_handle_cache_value_consistency() {
        let vec1 = "item1".as_bytes().to_vec();
        let vec2 = "item2".as_bytes().to_vec();
        assert_eq!(
            handle_cache_value_consistency(vec1, vec2.clone()).unwrap(),
            vec2
        );
    }

    #[test]
    fn test_handle_cache_value_consistency_when_remove_or_evict() {
        let vec1 = "item1".as_bytes().to_vec();
        let vec2 = "".as_bytes().to_vec();
        assert_eq!(
            handle_cache_value_consistency(vec1, vec2.clone()).unwrap(),
            vec2
        );
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 4);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"B".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_repeated_keys() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "D");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"B".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_less_keys() {
        let fifo1 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_more_keys() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "C".to_string(),
            "D".to_string(),
            "E".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "F".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "F");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 5);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"D".to_string()));
        assert!(merged_internal_data.fifo.contains(&"E".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"F".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "F".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_insert_with_initial_more_keys_and_repeated() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "C".to_string(),
            "D".to_string(),
            "E".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "B".to_string(), "D".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "D");

        let merged_state = handle_cache_state_consistency_when_insert(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 4);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"E".to_string()));
        assert_eq!(merged_internal_data.fifo.back().unwrap(), &"D".to_string());
        assert_eq!(merged_internal_data.last_modified_item_key, "D".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_remove_or_evict() {
        let fifo1 = VecDeque::from(["A".to_string(), "B".to_string(), "C".to_string()]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "B");

        let merged_state = handle_cache_state_consistency_when_remove_or_evict(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 2);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert_eq!(merged_internal_data.last_modified_item_key, "B".to_string());
    }

    #[test]
    fn test_handle_cache_state_consistency_when_remove_or_evict_with_prev_insert() {
        let fifo1 = VecDeque::from([
            "A".to_string(),
            "B".to_string(),
            "C".to_string(),
            "D".to_string(),
        ]);
        let state1: InternalData = InternalData::new(3, fifo1, "C");

        let fifo2 = VecDeque::from(["A".to_string(), "C".to_string()]);
        let state2: InternalData = InternalData::new(3, fifo2, "B");

        let merged_state = handle_cache_state_consistency_when_remove_or_evict(state1, state2);
        assert!(merged_state.is_some());

        let merged_internal_data = merged_state.unwrap();
        assert_eq!(merged_internal_data.fifo.len(), 3);
        assert!(merged_internal_data.fifo.contains(&"A".to_string()));
        assert!(merged_internal_data.fifo.contains(&"C".to_string()));
        assert!(merged_internal_data.fifo.contains(&"D".to_string()));
        assert_eq!(merged_internal_data.last_modified_item_key, "B".to_string());
    }

    #[test]
    fn remove_from_cache_when_item_exists() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-key1"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn remove_from_cache_when_item_is_deleted() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| None);

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(0)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn remove_from_cache_when_item_is_deleted_but_it_is_already_removed_from_state() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-keyA"),
            String::from("test-cache-keyB"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-keyB");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(0)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Complete, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    // TODO: W-10712428 Update test when implement rollback for errors when save internal data.
    #[test]
    fn remove_from_cache_when_shared_data_fails_updating_state() {
        let stored_fifo = VecDeque::from([String::from("test-cache-key1")]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key1");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_insert()
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x, value: InternalData, _handle_consistency| {
                (TransactionStatus::Rejected, value)
            });

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.delete("key1");
    }

    #[test]
    fn get_item_from_cache_ok() {
        let obj = SerializableObject {
            property_one: 1,
            property_two: 10,
        };
        let serialized_obj = bincode::serialize(&obj).unwrap();

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .returning(move |_x: &str| Some(serialized_obj.clone()));

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);

        let value = cache.get("key1");
        assert!(value.is_some());
        assert_eq!(value.unwrap(), bincode::serialize(&obj).unwrap());
    }

    #[test]
    fn get_item_from_cache_empty() {
        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();

        mock_concurrent_shared_data
            .expect_get::<Vec<u8>>()
            .with(eq("test-cache-key1"))
            .times(1)
            .returning(move |_x: &str| None);

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        assert!(cache.get("key1").is_none());
    }

    #[test]
    fn test_purge() {
        let stored_fifo = VecDeque::from([
            String::from("test-cache-key1"),
            String::from("test-cache-key2"),
        ]);
        let stored_internal_data = InternalData::new(3, stored_fifo, "test-cache-key2");

        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| Some(stored_internal_data.clone()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key1"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-key2"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(Vec::<u8>::new()));

        mock_concurrent_shared_data
            .expect_remove()
            .times(1)
            .with(eq("test-cache-internal_data"))
            .in_sequence(&mut sequence)
            .returning(move |_key| Some(InternalData::new(3, VecDeque::new(), "test-cache-key2")));

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.purge();
    }

    #[test]
    fn test_purge_cache_empty() {
        let mut mock_concurrent_shared_data = MockConcurrentSharedData::default();
        let mut sequence = Sequence::new();

        mock_concurrent_shared_data
            .expect_get::<InternalData>()
            .with(eq("test-cache-internal_data"))
            .times(1)
            .in_sequence(&mut sequence)
            .returning(move |_x: &str| None);

        mock_concurrent_shared_data
            .expect_remove::<Vec<u8>>()
            .times(0)
            .in_sequence(&mut sequence);

        mock_concurrent_shared_data
            .expect_remove::<InternalData>()
            .times(0)
            .in_sequence(&mut sequence);

        let cache = FifoCache::new(String::from("test-cache"), 3, mock_concurrent_shared_data);
        cache.purge();
    }
}