salrucc 0.13.0

use std::any::Any;
use std::borrow::Borrow;
use std::cmp::Eq;
use std::collections::HashMap;
use std::future::Future;
use std::hash::Hash;
use std::ops::DerefMut;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use tokio::sync::Mutex;
use tokio::sync::RwLock;
use tokio::sync::RwLockReadGuard;

use crate::queues::CacheQueues;
use crate::queues::ExpiryOption;
use crate::CacheClock;

/// The result of the retrieval future
pub struct RetrievalResult<V, D> {
	/// The value that gets cached
	pub value: V,

	/// The size of the footprint within the cache
	pub footprint: usize,

	/// Time to stale, or 'None' if it is never stale
	pub stale_expiry: Option<D>,

	/// Time to live, or 'None' if it lives forever
	pub ultimate_expiry: Option<D>,
}

/// Represents a single item in the cache's HashMap
struct EntryItem<C>
where
	C: CacheClock,
{
	/// The actual value, guarded by a mutex
	pub value: Mutex<Option<EntryItemValue<C>>>,

	/// An optional stale value, guarded with a RwLock
	pub stale: RwLock<Option<StaleItem<C>>>,
}

struct StaleItem<C>
where
	C: CacheClock,
{
	pub value: ItemBox,
	pub absolute_ultimate_expiry: Option<C::Instant>,
}

struct EntryItemValue<C>
where
	C: CacheClock,
{
	pub item: ItemBox,
	pub footprint: usize,
	pub absolute_stale_expiry: Option<C::Instant>,
	pub absolute_ultimate_expiry: Option<C::Instant>,
}

type ItemBox = Box<dyn Any + Send + Sync>;
type Entry<C> = Arc<EntryItem<C>>;
type Map<K, C> = HashMap<K, Entry<C>>;

/// A key-value LRU cache.
pub struct Cache<K, C>
where
	K: Clone + Hash + Eq + Send + Sync + 'static,
	C: CacheClock,
{
	/// The actual `HashMap` that contains the actual cached data.  Each item in the cache is in turn
	/// guarded by a `RwLock` specific to each entry.  The `HashMap` is itself guarded by an `RwLock`
	/// that protects the hash itself, but not the individual items
	cache: RwLock<Map<K, C>>,
	queues: Mutex<CacheQueues<K, C::Instant>>,
	cache_size: AtomicUsize,
	max_cache_size: usize,
	clock: C,
}

const CACHE_ENTRY_MISSING: &str = "cache value unexpectedly missing";

enum PopType {
	LeastRecentlyUsed,
	Expired,
}

#[derive(Debug, PartialEq)]
enum PostTask {
	None,
	SpawnMaintainCacheSize,
}

impl<K, C: CacheClock> Cache<K, C>
where
	K: Clone + Hash + Eq + Send + Sync + 'static,
{
	pub fn new(clock: C, max_cache_size: usize) -> Arc<Self> {
		Arc::new(Self {
			cache: RwLock::new(HashMap::new()),
			queues: Mutex::new(CacheQueues::new()),
			cache_size: AtomicUsize::new(0),
			max_cache_size,
			clock,
		})
	}

	/// Get a value from the cache, or generate it with the specified future.
	pub async fn get_or_insert_with<Q, V, E>(
		self: &Arc<Self>,
		key: &Q,
		future: impl Future<Output = Result<RetrievalResult<V, C::Duration>, E>>,
	) -> Result<V, E>
	where
		for<'a> K: Borrow<Q> + From<&'a Q>,
		Q: Hash + Eq + ?Sized,
		V: Clone + Send + Sync + 'static,
	{
		// Do the heavy lifting
		self.internal_get_or_insert_with(key, future, async {}, async {})
			.await
			.map(|(value, post_task)| {
				// When successfully invoked, we might have a "post task" to spawn something to maintain the size of the cache
				match post_task {
					PostTask::None => {}
					PostTask::SpawnMaintainCacheSize => {
						// Looks like we need to spawn a new thread to maintain the cache size
						let cache = Arc::clone(self);
						tokio::spawn(async move { cache.maintain_cache_size().await });
					}
				};
				value
			})
	}

	/// Core implementation of get_or_insert_with(), in a separate function to facilitate unit testing (no non-deterministic
	/// task spawning in unit tests!)
	async fn internal_get_or_insert_with<Q, V, E>(
		self: &Arc<Self>,
		key: &Q,
		future: impl Future<Output = Result<RetrievalResult<V, C::Duration>, E>>,
		midflight_future: impl Future<Output = ()>,
		queues_lock_future: impl Future<Output = ()>,
	) -> Result<(V, PostTask), E>
	where
		for<'a> K: Borrow<Q> + From<&'a Q>,
		Q: Hash + Eq + ?Sized,
		V: Clone + Send + Sync + 'static,
	{
		// First purge whatever might be expired
		self.purge_expired_items().await;

		// Access and lock the mutex for this specific key's cache entry - the lock for this key is held until the end of the function
		let (key, entry) = self.get_cache_entry(key).await;

		// There might be a stale value - try to get it as it will influence how we go about things
		let stale_value = entry
			.stale
			.read()
			.await
			.as_ref()
			.map(|x| {
				let is_expired = x
					.absolute_ultimate_expiry
					.as_ref()
					.is_some_and(|x| *x <= self.clock.now());
				(!is_expired)
					.then(|| x.value.downcast_ref::<V>().cloned())
					.flatten()
			})
			.flatten();

		let (item_exists, result, absolute_expiry, post_task) = {
			// At this point we need to lock the value mutex - note that if we have a
			// stale value, we have the option of crapping out and returning that
			// value.  For this reason, we're using `Result` with the stale value
			// as the `Err` value
			let guard_result = if let Some(stale_value) = stale_value {
				entry.value.try_lock().map_err(|_| stale_value)
			} else {
				Ok(entry.value.lock().await)
			};
			match guard_result {
				Err(stale_value) => {
					// Sort of happy path - we're returning a stale value
					(true, stale_value, ExpiryOption::Ignore, PostTask::None)
				}

				Ok(mut value_guard) => {
					// We have a lock on the value.  The first thing we need to do is check if
					// the value is stale and if so, relegate it to the stale field
					let is_stale = value_guard.as_ref().is_some_and(|value| {
						value
							.absolute_stale_expiry
							.as_ref()
							.is_some_and(|x| *x < self.clock.now())
					});
					if is_stale {
						let absolute_ultimate_expiry = value_guard
							.as_ref()
							.and_then(|x| x.absolute_ultimate_expiry.clone());
						let newly_stale_value = self.take_entry_item_value(value_guard.deref_mut());
						let newly_stale_item = newly_stale_value.map(|value| StaleItem {
							value,
							absolute_ultimate_expiry,
						});
						*entry.stale.write().await = newly_stale_item;
					}

					// Access the midflight future
					midflight_future.await;

					// Did we achieve a cache hit?
					let value = value_guard
						.as_ref()
						.and_then(|x| x.item.downcast_ref::<V>())
						.cloned();
					if let Some(value) = value {
						// We did - this is of course the (real) happy path
						(true, value, ExpiryOption::Ignore, PostTask::None)
					} else {
						// We had a cache miss, so we need to invoke `future`; note that we're only calling `future` while we are holding onto the mutex, so
						// anyone else trying to get it will have to wait until we're done, at which point it should be populated unless we hit an error
						let retrieval_result = future.await?;

						// The retrieval result returns stale_expiry as a Duration.  We need an Instant.
						let absolute_stale_expiry =
							retrieval_result.stale_expiry.map(|x| self.clock.now() + x);
						let absolute_ultimate_expiry = retrieval_result
							.ultimate_expiry
							.map(|x| self.clock.now() + x);

						// We've retrieved the result asynchronously; the next step is to set it up in the cache.  But
						// first we need to wipe the cache entry (the next line is critical, because if there is an object
						// that is not of type `V`, it is essential that the footprint be deducted
						self.take_entry_item_value(value_guard.deref_mut());

						// And with that out of the say, put the newly retrieved item in the cache
						*value_guard = Some(EntryItemValue {
							item: Box::new(retrieval_result.value.clone()),
							footprint: retrieval_result.footprint,
							absolute_stale_expiry,
							absolute_ultimate_expiry: absolute_ultimate_expiry.clone(),
						});

						// Clear out any stale item
						*entry.stale.write().await = None;

						// Return an ExpiryOption for absolute_ultimate_expiry
						let absolute_ultimate_expiry = match absolute_ultimate_expiry {
							None => ExpiryOption::Never,
							Some(x) => ExpiryOption::Expires(x),
						};

						// Record the increased footprint; note two things:
						// - `fetch_add` returns the previous value, so we need to add `retrieval_result.footprint` to it to get the new cache size
						// - We may need to spawn a thread to perform evictions, but we want to let the caller drive this as the
						//   entire reason we have an "internal" method is to make this behavior unit testable
						let new_cache_size = self
							.cache_size
							.fetch_add(retrieval_result.footprint, Ordering::Relaxed)
							+ retrieval_result.footprint;
						let post_task = if new_cache_size > self.max_cache_size {
							PostTask::SpawnMaintainCacheSize
						} else {
							PostTask::None
						};

						// And finally return it
						(
							is_stale,
							retrieval_result.value,
							absolute_ultimate_expiry,
							post_task,
						)
					}
				}
			}
		};

		let mut queues = self.queues.lock().await;
		queues_lock_future.await;
		// If item exists in cache, but it doesnt exist in queues, this means its going
		// to be popped off soon as part of a purge process.
		// So we dont add it to the queue to make sure it doesnt exist in queues
		// without being present in cache.
		if !item_exists || queues.exists(&key) {
			// Access the key and optionally update ultimate_expiry, and we're done
			// Insert a key into the priority queue, with the lowest priority. If the
			// key was already present in the queue, its priority will be changed to
			// the lowest priority.
			queues.push(key, absolute_expiry);
		}
		Ok((result, post_task))
	}

	async fn get_cache_entry<Q>(&self, key: &Q) -> (K, Entry<C>)
	where
		for<'a> K: Borrow<Q> + From<&'a Q>,
		Q: Hash + Eq + ?Sized,
	{
		let reader = self.insert_key(key).await;
		// we can unwrap here because we never remove entries from the cache, and
		// we've just inserted the new entry if it was missing
		let (key, cache_value) = reader.get_key_value(key).expect(CACHE_ENTRY_MISSING);
		(key.clone(), Arc::clone(cache_value))
	}

	async fn insert_key<Q>(&self, key: &Q) -> RwLockReadGuard<'_, Map<K, C>>
	where
		for<'a> K: Borrow<Q> + From<&'a Q>,
		Q: Hash + Eq + ?Sized,
	{
		let mut reader = self.cache.read().await;
		if !reader.contains_key(key) {
			drop(reader);
			let entry_item = EntryItem {
				value: Mutex::new(None),
				stale: RwLock::new(None),
			};
			self.cache
				.write()
				.await
				.entry(key.into())
				.or_insert_with(|| Arc::new(entry_item));
			reader = self.cache.read().await;
		}
		reader
	}

	async fn maintain_cache_size(&self) {
		while self.maintain_cache_size_once(async {}).await {}
	}

	async fn maintain_cache_size_once(&self, queue_pop_future: impl Future<Output = ()>) -> bool {
		let exceeded_max_cache_size = self.cache_size.load(Ordering::Relaxed) > self.max_cache_size;
		if exceeded_max_cache_size {
			self.pop_from_queue_and_remove_from_cache(PopType::LeastRecentlyUsed, queue_pop_future)
				.await;
		}
		exceeded_max_cache_size
	}

	async fn purge_expired_items(&self) {
		while self
			.pop_from_queue_and_remove_from_cache(PopType::Expired, async {})
			.await
		{}
	}

	/// Try to remove the least recently used `T` or whatever expired, based on pop_type
	///
	/// The return value indicate whether an item was removed.
	///
	/// Note that this function does not remove the actual entry from the
	/// underlying [`HashMap`], since [`get_or_insert_with`] assumes
	/// that once an entry is created, it is never removed. Instead, it
	/// replaces the value of the entry with `None`.
	///
	/// [`get_or_insert_with`]: Self::get_or_insert_with
	async fn pop_from_queue_and_remove_from_cache(
		&self,
		pop_type: PopType,
		queue_pop_future: impl Future<Output = ()>,
	) -> bool {
		// Access the cache object first (with a read lock)
		let cache = self.cache.read().await;

		// The methods to pop off of the queue take a callback so we can acquire a lock
		// at the same time.  If the attempt to lock fails, we don't want to pop off the queue
		let try_lock_callback = |key: &K| {
			cache
				.get(key.borrow())
				.expect(CACHE_ENTRY_MISSING)
				.value
				.try_lock()
				.ok()
		};

		// Try to pop an item
		let mut popped_item = {
			let mut queues = self.queues.lock().await;
			match pop_type {
				PopType::LeastRecentlyUsed => queues.pop_least_recently_used(try_lock_callback),
				PopType::Expired => queues.pop_expired(&self.clock.now(), try_lock_callback),
			}
		};

		// Did we pop anything?
		if let Some(ref mut value_guard) = popped_item.as_mut() {
			// Hook to facilitate obnoxious timing issues in unit tests
			queue_pop_future.await;

			self.take_entry_item_value(value_guard);
		}
		popped_item.is_some()
	}

	/// Takes the ItemBox out of the entry and adjusts the cache size
	fn take_entry_item_value(&self, value: &mut Option<EntryItemValue<C>>) -> Option<ItemBox> {
		value.take().map(|entry| {
			self.cache_size
				.fetch_sub(entry.footprint, Ordering::Relaxed);
			entry.item
		})
	}
}

#[cfg(test)]
mod tests {
	use std::pin::pin;
	use std::result::Result;
	use std::sync::atomic::AtomicI32;
	use std::sync::atomic::Ordering;
	use std::sync::Arc;
	use std::task::Context;
	use std::time::Duration;

	use futures::future::join_all;
	use futures::Future;
	use rand::rngs::StdRng;
	use rand::Rng;
	use rand::SeedableRng;
	use std::task::Poll;
	use tokio::task::spawn;
	use tokio::task::yield_now;

	use super::Cache;
	use super::PostTask;
	use super::RetrievalResult;

	use crate::fake_clock::FakeClock;
	use crate::fake_clock::FakeClockDuration;

	async fn rr<T>(value: T) -> Result<RetrievalResult<T, FakeClockDuration>, ()> {
		Ok(RetrievalResult {
			value,
			footprint: 0,
			ultimate_expiry: None,
			stale_expiry: None,
		})
	}

	async fn rr_fp(value: usize) -> Result<RetrievalResult<usize, FakeClockDuration>, ()> {
		Ok(RetrievalResult {
			value,
			footprint: value,
			ultimate_expiry: None,
			stale_expiry: None,
		})
	}

	async fn rr_expiry<T>(
		value: T,
		ultimate_expiry: FakeClockDuration,
	) -> Result<RetrievalResult<T, FakeClockDuration>, ()> {
		Ok(RetrievalResult {
			value,
			footprint: 0,
			ultimate_expiry: Some(ultimate_expiry),
			stale_expiry: None,
		})
	}

	async fn rr_unexpected<T>() -> Result<RetrievalResult<T, FakeClockDuration>, ()> {
		panic!("Should not get here")
	}

	#[tokio::test]
	async fn key_str() {
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), 100);
		assert_eq!(Ok(100), cache.get_or_insert_with("a", rr_fp(100)).await);
	}

	#[tokio::test]
	async fn key_i32array() {
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<[i32]>, _>::new(fake_clock, 100);
		let key = [123i32, 456, 789];
		let actual = cache.get_or_insert_with(&key[..], rr_fp(100)).await;
		assert_eq!(Ok(100), actual);
	}

	#[tokio::test]
	async fn double_loading() {
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock, 100);
		let t1 = cache.get_or_insert_with("a", async {
			tokio::time::sleep(Duration::from_secs(1)).await;
			rr_fp(1).await
		});
		let t2 = cache.get_or_insert_with("a", rr_unexpected::<usize>());
		let (res1, res2) = futures::join!(t1, t2);
		assert_eq!((res1.unwrap(), res2.unwrap()), (1, 1));
	}

	#[tokio::test]
	async fn heterogeneous() {
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock, 100);

		assert_eq!(Ok(42), cache.get_or_insert_with("a", rr(42)).await);
		assert_eq!(Ok('X'), cache.get_or_insert_with("b", rr('X')).await);
		assert_eq!(Ok('Y'), cache.get_or_insert_with("a", rr('Y')).await);
		assert_eq!(Ok('Y'), cache.get_or_insert_with("a", rr('Z')).await);
	}

	#[tokio::test]
	async fn eviction() {
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock, 250);

		assert_eq!(Ok(100), cache.get_or_insert_with("a", rr_fp(100)).await,);
		assert_eq!(Ok(100), cache.get_or_insert_with("b", rr_fp(100)).await,);
		assert_eq!(
			Ok(100),
			cache
				.get_or_insert_with("a", rr_unexpected::<usize>())
				.await,
		);

		// "b" should be evicted, since it is less recently accessed than "a"
		assert_eq!(Ok(100), cache.get_or_insert_with("c", rr_fp(100)).await,);
		// wait a second to make sure the cache eviction thread has time to do its thing
		tokio::time::sleep(std::time::Duration::from_secs(1)).await;
		// still no new request for "a"
		assert_eq!(
			Ok(100),
			cache
				.get_or_insert_with("a", rr_unexpected::<usize>())
				.await,
		);
		// new request for "b" since it was evicted after request for "c"
		assert_eq!(
			Ok(42), // if it wasn't evicted, we'd still see 100
			cache.get_or_insert_with("b", rr_fp(42)).await,
		);
	}

	#[tokio::test]
	async fn once() {
		let cache = Cache::<Arc<str>, _>::new(FakeClock::new(), 10000);
		assert_eq!(cache.get_or_insert_with("a", rr('X')).await.unwrap(), 'X');
		assert_eq!(
			Ok('X'),
			cache.get_or_insert_with("a", rr_unexpected::<char>()).await,
		);
	}

	#[tokio::test]
	async fn ultimate_expiry() {
		let fake_clock = FakeClock::new();
		fake_clock.set_time(0);
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), 10000);

		assert_eq!(Ok('A'), cache.get_or_insert_with("a", rr('A')).await);
		assert_eq!(
			Ok('B'),
			cache.get_or_insert_with("b", rr_expiry('B', 10)).await
		);
		assert_eq!(
			Ok('C'),
			cache.get_or_insert_with("c", rr_expiry('C', 20)).await
		);
		assert_eq!(
			Ok('A'),
			cache.get_or_insert_with("a", rr_unexpected::<char>()).await,
		);
		assert_eq!(
			Ok('B'),
			cache.get_or_insert_with("b", rr_unexpected::<char>()).await,
		);
		assert_eq!(
			Ok('C'),
			cache.get_or_insert_with("c", rr_unexpected::<char>()).await
		);

		// Set the clock to expire "b", but not "a" or "c"
		fake_clock.set_time(15);
		assert_eq!(
			Ok('A'),
			cache.get_or_insert_with("a", rr_unexpected::<char>()).await,
		);
		assert_eq!(
			Ok('X'),
			cache.get_or_insert_with("b", rr_expiry('X', 10)).await
		);
		assert_eq!(
			Ok('C'),
			cache.get_or_insert_with("c", rr_unexpected::<char>()).await,
		);
	}

	#[tokio::test]
	pub async fn stale() {
		// Create the cache with a fake clock
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), 10000);

		// A specialized fake retrieve function, with a retrieval count
		let return_value = Arc::new(AtomicI32::new(100));
		let my_retrieve = || async {
			tokio::time::sleep(Duration::from_millis(1)).await;
			let value = return_value.fetch_add(1, Ordering::Relaxed);
			let result: Result<_, ()> = Ok(RetrievalResult {
				value,
				footprint: 100,
				stale_expiry: Some(10),
				ultimate_expiry: Some(20),
			});
			result
		};

		// Access the cache; this forces a retrieve
		let actual = cache.get_or_insert_with("foo", my_retrieve()).await;
		assert_eq!(Ok(100), actual);

		// Repeat a few times
		let actual = cache.get_or_insert_with("foo", my_retrieve()).await;
		assert_eq!(Ok(100), actual);
		let actual = cache.get_or_insert_with("foo", my_retrieve()).await;
		assert_eq!(Ok(100), actual);

		// Advance to clock by 15, and issue two requests -one of them should return a live
		// value and the other should return a stale value
		fake_clock.set_time(15);
		let results = join_all([
			cache.get_or_insert_with("foo", my_retrieve()),
			cache.get_or_insert_with("foo", my_retrieve()),
			cache.get_or_insert_with("foo", my_retrieve()),
		])
		.await;
		assert_eq!([Ok(101), Ok(100), Ok(100)], results.as_ref());

		// Now advance to 50 and reissue the requests; this should cause all of them
		// to hit the ultimate expiry
		fake_clock.set_time(50);
		let results = join_all([
			cache.get_or_insert_with("foo", my_retrieve()),
			cache.get_or_insert_with("foo", my_retrieve()),
			cache.get_or_insert_with("foo", my_retrieve()),
		])
		.await;
		assert_eq!([Ok(102), Ok(102), Ok(102)], results.as_ref());
	}

	#[tokio::test]
	pub async fn lots_of_activity() {
		// Constants
		const TICK_COUNT: i32 = 2048;
		const HITS_PER_TICK: i32 = 32;
		const RANDOM_SEED: u64 = 31337;
		const CACHE_SIZE: usize = 100000;
		const ITEM_MAX_SIZE: usize = 40000;

		// Create the cache with a fake clock
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), CACHE_SIZE);

		// Instantiate a random number generator
		let mut random = StdRng::seed_from_u64(RANDOM_SEED);

		for tick in 0..TICK_COUNT {
			fake_clock.set_time(tick);

			let tasks = (0..HITS_PER_TICK).map(|_| {
				// Decide the key/value/ultimate_expiry/footprint for this request using deterministic (i.e. - seeded) random number
				let random_value = random.gen::<usize>() % ITEM_MAX_SIZE;
				let key = format!("key{random_value}");
				let value = format!("value{random_value}");
				let stale_expiry =
					(random_value & 0x100 != 0).then_some((random_value % 100) as i32);
				let ultimate_expiry = stale_expiry.map(|x| x + 10);
				let footprint = random_value % 1000 + 10;
				let do_yield = random_value & 0x200 != 0;

				// Build the RetrievalResult
				let retrieval_result = RetrievalResult {
					value: value.clone(),
					footprint,
					ultimate_expiry,
					stale_expiry,
				};

				// And launch the task
				let cache = cache.clone();
				let task = async move {
					let result: Result<String, ()> = cache
						.get_or_insert_with(key.as_str(), async {
							if do_yield {
								yield_now().await;
							}
							Ok(retrieval_result)
						})
						.await;
					(value, result)
				};
				spawn(task)
			});
			for spawn_result in join_all(tasks).await {
				let Ok((value, result)) = spawn_result else {
					panic!("Spawn failed: {spawn_result:?}");
				};
				assert_eq!(Ok(value), result);
			}
		}
	}

	#[tokio::test]
	async fn snowflake_timing_issue1() {
		// Step #1 - Create the cache
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), 100);
		let cx = &mut Context::from_waker(&futures::task::noop_waker_ref());

		// Step #2 - Add a resource that is expected to overflow the cache
		let result = cache
			.internal_get_or_insert_with("foo", rr_fp(200), async {}, async {})
			.await;
		assert_eq!(Ok((200, PostTask::SpawnMaintainCacheSize)), result);

		// Step #3 - Access the exact same resource again in rapid succession; we want a scenario where we
		// reacquire the lock before maintaining the cache size - BUT once in the lock, we're going to wait
		// until time T+10
		let mut access_future =
			pin!(cache.internal_get_or_insert_with("foo", rr_fp(200), fake_clock.wait_until(10), async {}));
		assert_eq!(Poll::Pending, access_future.as_mut().poll(cx));

		// Step #3 - In step #2 we an item that overflowed the cache; kick off the process of maintaining the
		// cache size - BUT this won't do anything because it cannot access the item's lock
		let mut maintain_cache_size_future =
			pin!(cache.maintain_cache_size_once(async { unreachable!() }));
		assert_eq!(
			Poll::Ready(true),
			maintain_cache_size_future.as_mut().poll(cx)
		);

		// Step #4 - Advance to T+10; the call to maintain_cache_size_once should finish up but the other call should
		// still be pending
		fake_clock.set_time(10);
		assert_eq!(
			Poll::Ready(Ok((200, PostTask::None))),
			access_future.as_mut().poll(cx)
		);

		// Step #5 - Advance to T+50; let the process of cache size maintenance complete
		fake_clock.set_time(50);
		assert!(cache.maintain_cache_size_once(async {}).await);

		// Step #6 - Verify that we're empty once again
		let cache_size = cache.cache_size.load(Ordering::Relaxed);
		let cache_queues_are_empty = cache.queues.lock().await.is_empty();
		assert_eq!((0, true), (cache_size, cache_queues_are_empty));
	}

	#[tokio::test]
	async fn snowflake_timing_issue2() {
		// Step #1 - Create the cache
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), usize::MAX);
		let cx = &mut Context::from_waker(&futures::task::noop_waker_ref());

		// Step #2 - Add a resource with distinct stale and ultimate expiries
		let retrieval_result = RetrievalResult {
			value: 1234,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("foo", async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);

		// Step #4 - Access that resource after it becomes stale, but before it ultimately expires BUT
		// the request to retrieve the item does not conclude until after the ultimate expiry.  Because
		// this does not resolve until later, the future will be pending
		fake_clock.set_time(190);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let mut access_future = pin!(cache.internal_get_or_insert_with(
			"foo",
			async {
				fake_clock.wait_until(220).await;
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {}
		));
		assert_eq!(Poll::Pending, access_future.as_mut().poll(cx));

		// Step #5 - After the first item completely expires, access the cache in such a way that we
		// purge the expired items
		fake_clock.set_time(210);
		let retrieval_result = RetrievalResult {
			value: 3456,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("not_foo", async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((3456, PostTask::None)), result);
		assert_eq!(Poll::Pending, access_future.as_mut().poll(cx));

		// Step #6 - Only now advance to T+220 and finish accessing `foo`
		fake_clock.set_time(220);
		assert_eq!(
			Poll::Ready(Ok((2345, PostTask::None))),
			access_future.as_mut().poll(cx)
		);
	}

	#[tokio::test]
	async fn snowflake_timing_issue3() {
		// Step #1 - Create the cache
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), usize::MAX);
		let cx = &mut Context::from_waker(&futures::task::noop_waker_ref());

		// Step #2 - Add a resource with distinct stale and ultimate expiries
		let retrieval_result = RetrievalResult {
			value: 1234,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("foo", async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);

		// Step #3 - When this resource becomes stale, hit it again - and with a long lived request
		fake_clock.set_time(150);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let mut access_future1 = pin!(cache.internal_get_or_insert_with(
			"foo",
			async {
				fake_clock.wait_until(300).await;
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {}
		));
		assert_eq!(Poll::Pending, access_future1.as_mut().poll(cx));

		// Step #4 - Hit this resource again; we should be served a stale request
		let result = cache
			.internal_get_or_insert_with("foo", rr_unexpected(), async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);
		assert_eq!(Poll::Pending, access_future1.as_mut().poll(cx));

		// Step #5 - Advance to when the item in the cache has expired, but access_future1 has not completed
		fake_clock.set_time(250);
		let mut access_future2 =
			pin!(cache.internal_get_or_insert_with("foo", rr_unexpected(), async {}, async {}));
		assert_eq!(Poll::Pending, access_future1.as_mut().poll(cx));
		assert_eq!(Poll::Pending, access_future2.as_mut().poll(cx));

		// Step #6 - Finally advance to the time when the access futures complete
		fake_clock.set_time(300);
		assert_eq!(
			Poll::Ready(Ok((2345, PostTask::None))),
			access_future1.as_mut().poll(cx)
		);
		assert_eq!(
			Poll::Ready(Ok((2345, PostTask::None))),
			access_future2.as_mut().poll(cx)
		);
	}

	#[tokio::test]
	async fn snowflake_timing_issue4() {
		// Step #1 - Create the cache
		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), usize::MAX);
		let cx = &mut Context::from_waker(&futures::task::noop_waker_ref());

		// Step #2 - Add a resource with distinct stale and ultimate expiries
		let retrieval_result = RetrievalResult {
			value: 1234,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("foo", async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);

		// Step #3 - When this resource becomes stale, hit it again - and with a long lived request
		fake_clock.set_time(150);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let mut access_future1 = pin!(cache.internal_get_or_insert_with(
			"foo",
			async {
				fake_clock.wait_until(250).await;
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {},
		));
		assert_eq!(Poll::Pending, access_future1.as_mut().poll(cx));

		// Step #4 - We create a separate request for a different item, 
		// but the purpose of this is to hold the queue lock for some time.
		// This can happen in real situations, we try to replicate this using this behavior.
		// Currently, the below request will hold the queue lock till 300 time units.
		fake_clock.set_time(175);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(10000),
			ultimate_expiry: Some(20000),
		};
		let mut access_future2 = pin!(cache.internal_get_or_insert_with(
			"not foo",
			async {
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {
				fake_clock.wait_until(300).await;
			},
		));
		assert_eq!(Poll::Pending, access_future2.as_mut().poll(cx));

		// Step #5 - Trigger purge of "foo" using another request, since it expired at 200 seconds.
		// But this will not proceed as queue is locked by request in Step 3.
		fake_clock.set_time(210);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(10000),
			ultimate_expiry: Some(20000),
		};
		let mut access_future3 = pin!(cache.internal_get_or_insert_with(
			"not foo again",
			async {
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {},
		));
		assert_eq!(Poll::Pending, access_future3.as_mut().poll(cx));

		// Step #6 - We increase the time to 310
		fake_clock.set_time(310);

		// Step #7 - We allow Step 3 future to complete, but queue is still locked.
		assert_eq!(Poll::Pending, access_future1.as_mut().poll(cx));

		// Step #8 - We allow the queues lock to be released, since it was held till 300 seconds
		assert_eq!(Poll::Ready(Ok((2345, PostTask::None))), access_future2.as_mut().poll(cx));

		// Step #9 - Now will allow the purge process to get the queues lock, it will
		// also be able to get the cache value lock, and hence will purge the item.
		assert_eq!(Poll::Pending, access_future3.as_mut().poll(cx));

		// Step #10 - Now this will get the queue lock, and complete the request.
		assert_eq!(Poll::Ready(Ok((2345, PostTask::None))), access_future1.as_mut().poll(cx));

		// Step #11 - Finally we allow the request that initiated the purge will also complete.
		assert_eq!(Poll::Ready(Ok((2345, PostTask::None))), access_future3.as_mut().poll(cx));

		// We have now created a situation, the item "foo" was purged from the cache in Step 9
		// cache, but it was added in queue again by Step 10.

		// Step #10 - Now we reach a time where everything is expired, and triggering a request,
		// will purge expired items, without the current changes, this would cause a panic.
		fake_clock.set_time(800);
		let retrieval_result = RetrievalResult {
			value: 2345,
			footprint: 100,
			stale_expiry: Some(10000),
			ultimate_expiry: Some(20000),
		};
		let mut access_future4 = pin!(cache.internal_get_or_insert_with(
			"not foo again 2",
			async {
				let result: Result<_, ()> = Ok(retrieval_result);
				result
			},
			async {},
			async {},
		));
		assert_eq!(Poll::Ready(Ok((2345, PostTask::None))), access_future4.as_mut().poll(cx));
	}

	#[tokio::test]
	async fn snowflake_timing_issue5() {
		// We want to test a scenario where item is stale and so it is removed from the value in EntryItem, but added to stale.
		// But then the retrieval fails, so the item is not there in value, but still available in queue.

		let fake_clock = FakeClock::new();
		let cache = Cache::<Arc<str>, _>::new(fake_clock.clone(), usize::MAX);

		// Step #2 - Add a resource with distinct stale and ultimate expiries
		let retrieval_result = RetrievalResult {
			value: 1234,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("foo", async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);


		// Step #3 - When this resource becomes stale, hit it again but the retrieval results in an error, 
		// so we dont make any changes in the queues, but the item is available as stale.
		fake_clock.set_time(150);
		let result: Result<((), PostTask), &'static str> = cache
			.internal_get_or_insert_with("foo", async { Err("Some error occured") }, async {}, async {})
			.await;
		assert_eq!(Err("Some error occured"), result);

		// Step #4 - When the first resource becomes expired now, we try to request another item. 
		// This should throw an error without the fix because the queues will pop the first item, but the cache value will be missing
		fake_clock.set_time(250);
		let retrieval_result = RetrievalResult {
			value: 1234,
			footprint: 100,
			stale_expiry: Some(100),
			ultimate_expiry: Some(200),
		};
		let result: Result<_, ()> = cache
			.internal_get_or_insert_with("not foo",  async { Ok(retrieval_result) }, async {}, async {})
			.await;
		assert_eq!(Ok((1234, PostTask::None)), result);
	}
}