sequential_storage/queue/
mod.rs

1//! Implementation of the queue logic.
2
3pub mod buffered;
4
5use crate::item::{Item, ItemHeader, ItemHeaderIter};
6
7use self::{cache::CacheImpl, item::ItemUnborrowed};
8
9use super::{
10    Debug, Deref, DerefMut, Error, GenericStorage, MAX_WORD_SIZE, NorFlash, NorFlashExt, PageState,
11    PhantomData, Range, cache, calculate_page_address, calculate_page_end_address,
12    calculate_page_index, calculate_page_size, item, run_with_auto_repair,
13};
14use embedded_storage_async::nor_flash::MultiwriteNorFlash;
15
16/// Configuration for a queue
17pub struct QueueConfig<S> {
18    flash_range: Range<u32>,
19    _phantom: PhantomData<S>,
20}
21
22impl<S: NorFlash> QueueConfig<S> {
23    /// Create a new queue configuration. Will panic if the data is invalid.
24    /// If you want a fallible version, use [`Self::try_new`].
25    #[must_use]
26    pub const fn new(flash_range: Range<u32>) -> Self {
27        match Self::try_new(flash_range) {
28            Ok(config) => config,
29            Err(_) => panic!("Queue config must be correct"),
30        }
31    }
32
33    /// Create a new queue configuration. Will return None if the data is invalid
34    pub const fn try_new(flash_range: Range<u32>) -> Result<Self, QueueConfigError> {
35        if !flash_range.start.is_multiple_of(S::ERASE_SIZE as u32) {
36            return Err(QueueConfigError::StartRangeNotAtPageBoundary);
37        }
38        if !flash_range.end.is_multiple_of(S::ERASE_SIZE as u32) {
39            return Err(QueueConfigError::EndRangeNotAtPageBoundary);
40        }
41        // At least 1 page is used
42        if flash_range.end - flash_range.start < (S::ERASE_SIZE as u32) {
43            return Err(QueueConfigError::RangeTooSmall);
44        }
45
46        // A page must be able to store the 2 page states and at least one item header and a word for item data
47        if S::ERASE_SIZE < S::WORD_SIZE * 3 + ItemHeader::data_address::<S>(0) as usize {
48            return Err(QueueConfigError::PagesTooSmall);
49        }
50        if S::WORD_SIZE > MAX_WORD_SIZE {
51            return Err(QueueConfigError::WordSizeTooLarge);
52        }
53
54        Ok(Self {
55            flash_range,
56            _phantom: PhantomData,
57        })
58    }
59}
60
61/// Error for [`QueueConfig`] constructor
62#[derive(Debug, Clone, PartialEq, Eq)]
63#[cfg_attr(feature = "defmt", derive(defmt::Format))]
64pub enum QueueConfigError {
65    /// The start range address is not a multiple of the erase size
66    StartRangeNotAtPageBoundary,
67    /// The end range address is not a multiple of the erase size
68    EndRangeNotAtPageBoundary,
69    /// The queue needs at least 1 page to operate, but got less
70    RangeTooSmall,
71    /// The pages of the flash are too small to work with
72    PagesTooSmall,
73    /// The word size of the flash is bigger than [`MAX_WORD_SIZE`]
74    WordSizeTooLarge,
75}
76
77/// A fifo queue storage
78///
79/// Use [`Self::push`] to add data to the fifo and use [`Self::peek`] and [`Self::pop`] to get the data back.
80///
81/// ## Basic API
82///
83/// ```rust
84/// # use sequential_storage::cache::Cache;
85/// # use sequential_storage::queue::{QueueConfig, QueueStorage};
86/// # use mock_flash::MockFlashBase;
87/// # use futures::executor::block_on;
88/// # type Flash = MockFlashBase<10, 1, 4096>;
89/// # mod mock_flash {
90/// #   include!("../mock_flash.rs");
91/// # }
92/// #
93/// # fn init_flash() -> Flash {
94/// #     Flash::new(mock_flash::WriteCountCheck::Twice, None, false)
95/// # }
96/// #
97/// # block_on(async {
98///
99/// // Initialize the flash. This can be internal or external
100/// let mut flash = init_flash();
101///
102/// let mut storage = QueueStorage::new(flash, const { QueueConfig::new(0x1000..0x3000) }, Cache::new_uncached());
103/// // We need to give the crate a buffer to work with.
104/// // It must be big enough to serialize the biggest value of your storage type in.
105/// let mut data_buffer = [0; 128];
106///
107/// let my_data = [10, 47, 29];
108///
109/// // We can push some data to the queue
110/// storage.push(&my_data, false).await.unwrap();
111///
112/// // We can peek at the oldest data
113///
114/// assert_eq!(
115///     &storage.peek(&mut data_buffer).await.unwrap().unwrap()[..],
116///     &my_data[..]
117/// );
118///
119/// // With popping we get back the oldest data, but that data is now also removed
120///
121/// assert_eq!(
122///     &storage.pop(&mut data_buffer).await.unwrap().unwrap()[..],
123///     &my_data[..]
124/// );
125///
126/// // If we pop again, we find there's no data anymore
127///
128/// assert_eq!(
129///     storage.pop(&mut data_buffer).await,
130///     Ok(None)
131/// );
132/// # });
133/// ```
134pub struct QueueStorage<S: NorFlash, C: CacheImpl<()>> {
135    inner: GenericStorage<S, C, ()>,
136}
137
138impl<S: NorFlash, C: CacheImpl<()>> QueueStorage<S, C> {
139    /// Create a new (fifo) queue instance
140    ///
141    /// The provided cache instance must be new or must be in the exact correct state for the current flash contents.
142    /// If the cache is bad, undesirable things will happen.
143    /// So, it's ok to reuse the cache gotten from the [`Self::destroy`] method when the flash hasn't changed since calling destroy.
144    pub const fn new(storage: S, config: QueueConfig<S>, cache: C) -> Self {
145        Self {
146            inner: GenericStorage {
147                flash: storage,
148                flash_range: config.flash_range,
149                cache,
150                _phantom: PhantomData,
151            },
152        }
153    }
154
155    /// Push data into the queue.
156    /// The data can only be taken out with the [`Self::pop`] function.
157    ///
158    /// Old data will not be overwritten unless `allow_overwrite_old_data` is true.
159    /// If it is, then if the queue is full, the oldest data is removed to make space for the new data.
160    ///
161    /// *Note: If a page is already used and you push more data than the remaining capacity of the page,
162    /// the entire remaining capacity will go unused because the data is stored on the next page.*
163    pub async fn push(
164        &mut self,
165        data: &[u8],
166        allow_overwrite_old_data: bool,
167    ) -> Result<(), Error<S::Error>> {
168        run_with_auto_repair!(
169            function = self.push_inner(data, allow_overwrite_old_data).await,
170            repair = self.try_repair().await?
171        )
172    }
173
174    async fn push_inner(
175        &mut self,
176        data: &[u8],
177        allow_overwrite_old_data: bool,
178    ) -> Result<(), Error<S::Error>> {
179        if self.inner.cache.is_dirty() {
180            self.inner.cache.invalidate_cache_state();
181        }
182
183        // Data must fit in a single page
184        if data.len() > u16::MAX as usize
185            || data.len()
186                > calculate_page_size::<S>()
187                    .saturating_sub(ItemHeader::data_address::<S>(0) as usize)
188        {
189            self.inner.cache.unmark_dirty();
190            return Err(Error::ItemTooBig);
191        }
192
193        let current_page = self.find_youngest_page().await?;
194
195        let page_data_start_address =
196            calculate_page_address::<S>(self.flash_range(), current_page) + S::WORD_SIZE as u32;
197        let page_data_end_address =
198            calculate_page_end_address::<S>(self.flash_range(), current_page) - S::WORD_SIZE as u32;
199
200        self.inner.partial_close_page(current_page).await?;
201
202        // Find the last item on the page so we know where we need to write
203
204        let mut next_address = self
205            .inner
206            .find_next_free_item_spot(
207                page_data_start_address,
208                page_data_end_address,
209                data.len() as u32,
210            )
211            .await?;
212
213        if next_address.is_none() {
214            // No cap left on this page, move to the next page
215            let next_page = self.inner.next_page(current_page);
216            let next_page_state = self.inner.get_page_state_cached(next_page).await?;
217            let single_page = next_page == current_page;
218
219            match (next_page_state, single_page) {
220                (PageState::Open, _) => {
221                    self.inner.close_page(current_page).await?;
222                    self.inner.partial_close_page(next_page).await?;
223                    next_address = Some(
224                        calculate_page_address::<S>(self.flash_range(), next_page)
225                            + S::WORD_SIZE as u32,
226                    );
227                }
228                (PageState::Closed, _) | (PageState::PartialOpen, true) => {
229                    let next_page_data_start_address =
230                        calculate_page_address::<S>(self.flash_range(), next_page)
231                            + S::WORD_SIZE as u32;
232
233                    if !allow_overwrite_old_data
234                        && self.inner.page_contains_live_data(next_page).await?
235                    {
236                        self.inner.cache.unmark_dirty();
237                        return Err(Error::FullStorage);
238                    }
239
240                    self.inner.open_page(next_page).await?;
241                    if !single_page {
242                        self.inner.close_page(current_page).await?;
243                    }
244                    self.inner.partial_close_page(next_page).await?;
245                    next_address = Some(next_page_data_start_address);
246                }
247                (PageState::PartialOpen, false) => {
248                    // This should never happen
249                    return Err(Error::Corrupted {
250                        #[cfg(feature = "_test")]
251                        backtrace: std::backtrace::Backtrace::capture(),
252                    });
253                }
254            }
255        }
256
257        Item::write_new(
258            &mut self.inner.flash,
259            self.inner.flash_range.clone(),
260            &mut self.inner.cache,
261            next_address.unwrap(),
262            data,
263        )
264        .await?;
265
266        self.inner.cache.unmark_dirty();
267        Ok(())
268    }
269
270    /// Get an iterator-like interface to iterate over the items stored in the queue.
271    /// This goes from oldest to newest.
272    ///
273    /// The iteration happens non-destructively, or in other words it peeks at every item.
274    /// The returned entry has a [`QueueIteratorEntry::pop`] function with which you can decide to pop the item
275    /// after you've seen the contents.
276    pub async fn iter(&mut self) -> Result<QueueIterator<'_, S, C>, Error<S::Error>> {
277        // Note: Corruption repair is done in these functions already
278        QueueIterator::new(self).await
279    }
280
281    /// Peek at the oldest data.
282    ///
283    /// If you also want to remove the data use [`Self::pop`].
284    ///
285    /// The data is written to the given `data_buffer` and the part that was written is returned.
286    /// It is valid to only use the length of the returned slice and use the original `data_buffer`.
287    /// The `data_buffer` may contain extra data on ranges after the returned slice.
288    /// You should not depend on that data.
289    ///
290    /// If the data buffer is not big enough an error is returned.
291    pub async fn peek<'d>(
292        &mut self,
293        data_buffer: &'d mut [u8],
294    ) -> Result<Option<&'d mut [u8]>, Error<S::Error>> {
295        // Note: Corruption repair is done in these functions already
296        let mut iterator = self.iter().await?;
297
298        let next_value = iterator.next(data_buffer).await?;
299
300        match next_value {
301            Some(entry) => Ok(Some(entry.into_buf())),
302            None => Ok(None),
303        }
304    }
305
306    /// Pop the oldest data from the queue.
307    ///
308    /// If you don't want to remove the data use [`Self::peek`].
309    ///
310    /// The data is written to the given `data_buffer` and the part that was written is returned.
311    /// It is valid to only use the length of the returned slice and use the original `data_buffer`.
312    /// The `data_buffer` may contain extra data on ranges after the returned slice.
313    /// You should not depend on that data.
314    ///
315    /// If the data buffer is not big enough an error is returned.
316    pub async fn pop<'d>(
317        &mut self,
318        data_buffer: &'d mut [u8],
319    ) -> Result<Option<&'d mut [u8]>, Error<S::Error>>
320    where
321        S: MultiwriteNorFlash,
322    {
323        let mut iterator = self.iter().await?;
324
325        let next_value = iterator.next(data_buffer).await?;
326
327        match next_value {
328            Some(entry) => Ok(Some(entry.pop().await?)),
329            None => Ok(None),
330        }
331    }
332
333    /// Find the largest size of data that can be stored.
334    ///
335    /// This will read through the entire flash to find the largest chunk of
336    /// data that can be stored, taking alignment requirements of the item into account.
337    ///
338    /// If there is no space left, `None` is returned.
339    pub async fn find_max_fit(&mut self) -> Result<Option<u32>, Error<S::Error>> {
340        run_with_auto_repair!(
341            function = self.find_max_fit_inner().await,
342            repair = self.try_repair().await?
343        )
344    }
345
346    async fn find_max_fit_inner(&mut self) -> Result<Option<u32>, Error<S::Error>> {
347        if self.inner.cache.is_dirty() {
348            self.inner.cache.invalidate_cache_state();
349        }
350
351        let current_page = self.find_youngest_page().await?;
352
353        // Check if we have space on the next page
354        let next_page = self.inner.next_page(current_page);
355        match self.inner.get_page_state_cached(next_page).await? {
356            state @ PageState::Closed => {
357                if self.inner.is_page_empty(next_page, Some(state)).await? {
358                    self.inner.cache.unmark_dirty();
359                    return Ok(Some((S::ERASE_SIZE - (2 * S::WORD_SIZE)) as u32));
360                }
361            }
362            PageState::Open => {
363                self.inner.cache.unmark_dirty();
364                return Ok(Some((S::ERASE_SIZE - (2 * S::WORD_SIZE)) as u32));
365            }
366            PageState::PartialOpen => {
367                // This should never happen
368                return Err(Error::Corrupted {
369                    #[cfg(feature = "_test")]
370                    backtrace: std::backtrace::Backtrace::capture(),
371                });
372            }
373        }
374
375        // See how much space we can find in the current page.
376        let page_data_start_address =
377            calculate_page_address::<S>(self.flash_range(), current_page) + S::WORD_SIZE as u32;
378        let page_data_end_address =
379            calculate_page_end_address::<S>(self.flash_range(), current_page) - S::WORD_SIZE as u32;
380
381        let next_item_address = match self.inner.cache.first_item_after_written(current_page) {
382            Some(next_item_address) => next_item_address,
383            None => {
384                ItemHeaderIter::new(
385                    self.inner
386                        .cache
387                        .first_item_after_erased(current_page)
388                        .unwrap_or(page_data_start_address),
389                    page_data_end_address,
390                )
391                .traverse(&mut self.inner.flash, |_, _| true)
392                .await?
393                .1
394            }
395        };
396
397        self.inner.cache.unmark_dirty();
398        Ok(ItemHeader::available_data_bytes::<S>(
399            page_data_end_address - next_item_address,
400        ))
401    }
402
403    /// Calculate how much space is left free in the queue (in bytes).
404    ///
405    /// The number given back is accurate, however there are lots of things that add overhead and padding.
406    /// Every push is an item with its own overhead. You can check the overhead per item with [`Self::item_overhead_size`].
407    ///
408    /// Furthermore, every item has to fully fit in a page. So if a page has 50 bytes left and you push an item of 60 bytes,
409    /// the current page is closed and the item is stored on the next page, 'wasting' the 50 you had.
410    ///
411    /// So unless you're tracking all this, the returned number should only be used as a rough indication.
412    pub async fn space_left(&mut self) -> Result<u32, Error<S::Error>> {
413        run_with_auto_repair!(
414            function = self.space_left_inner().await,
415            repair = self.try_repair().await?
416        )
417    }
418
419    async fn space_left_inner(&mut self) -> Result<u32, Error<S::Error>> {
420        if self.inner.cache.is_dirty() {
421            self.inner.cache.invalidate_cache_state();
422        }
423
424        let mut total_free_space = 0;
425
426        for page in self.inner.get_pages(0) {
427            let state = self.inner.get_page_state_cached(page).await?;
428            let page_empty = self.inner.is_page_empty(page, Some(state)).await?;
429
430            if state.is_closed() && !page_empty {
431                continue;
432            }
433
434            // See how much space we can find in the current page.
435            let page_data_start_address =
436                calculate_page_address::<S>(self.flash_range(), page) + S::WORD_SIZE as u32;
437            let page_data_end_address =
438                calculate_page_end_address::<S>(self.flash_range(), page) - S::WORD_SIZE as u32;
439
440            if page_empty {
441                total_free_space += page_data_end_address - page_data_start_address;
442                continue;
443            }
444
445            // Partial open page
446            let next_item_address = match self.inner.cache.first_item_after_written(page) {
447                Some(next_item_address) => next_item_address,
448                None => {
449                    ItemHeaderIter::new(
450                        self.inner
451                            .cache
452                            .first_item_after_erased(page)
453                            .unwrap_or(page_data_start_address),
454                        page_data_end_address,
455                    )
456                    .traverse(&mut self.inner.flash, |_, _| true)
457                    .await?
458                    .1
459                }
460            };
461
462            if ItemHeader::available_data_bytes::<S>(page_data_end_address - next_item_address)
463                .is_none()
464            {
465                // No data fits on this partial open page anymore.
466                // So if all data on this is already erased, then this page might as well be counted as empty.
467                // We can use [is_page_empty] and lie to to it so it checks the items.
468                if self
469                    .inner
470                    .is_page_empty(page, Some(PageState::Closed))
471                    .await?
472                {
473                    total_free_space += page_data_end_address - page_data_start_address;
474                    continue;
475                }
476            }
477
478            total_free_space += page_data_end_address - next_item_address;
479        }
480
481        self.inner.cache.unmark_dirty();
482        Ok(total_free_space)
483    }
484
485    async fn find_youngest_page(&mut self) -> Result<usize, Error<S::Error>> {
486        let last_used_page = self
487            .inner
488            .find_first_page(0, PageState::PartialOpen)
489            .await?;
490
491        if let Some(last_used_page) = last_used_page {
492            return Ok(last_used_page);
493        }
494
495        // We have no partial open page. Search for a closed page to anker ourselves to
496        let first_closed_page = self.inner.find_first_page(0, PageState::Closed).await?;
497
498        let first_open_page = match first_closed_page {
499            Some(anchor) => {
500                // We have at least one closed page
501                // The first one after is the page we need to use
502                self.inner.find_first_page(anchor, PageState::Open).await?
503            }
504            None => {
505                // No closed pages and no partial open pages, so all pages should be open
506                // Might as well start at page 0
507                Some(0)
508            }
509        };
510
511        if let Some(first_open_page) = first_open_page {
512            return Ok(first_open_page);
513        }
514
515        // All pages are closed... This is not correct.
516        Err(Error::Corrupted {
517            #[cfg(feature = "_test")]
518            backtrace: std::backtrace::Backtrace::capture(),
519        })
520    }
521
522    async fn find_oldest_page(&mut self) -> Result<usize, Error<S::Error>> {
523        let youngest_page = self.find_youngest_page().await?;
524
525        // The oldest page is the first non-open page after the youngest page
526        let oldest_closed_page = self
527            .inner
528            .find_first_page(youngest_page, PageState::Closed)
529            .await?;
530
531        Ok(oldest_closed_page.unwrap_or(youngest_page))
532    }
533
534    /// Try to repair the state of the flash to hopefully get back everything in working order.
535    /// Care is taken that no data is lost, but this depends on correctly repairing the state and
536    /// so is only best effort.
537    ///
538    /// This function might be called after a different function returned the [`Error::Corrupted`] error.
539    /// There's no guarantee it will work.
540    ///
541    /// If this function or the function call after this crate returns [`Error::Corrupted`], then it's unlikely
542    /// that the state can be recovered. To at least make everything function again at the cost of losing the data,
543    /// erase the flash range.
544    async fn try_repair(&mut self) -> Result<(), Error<S::Error>> {
545        self.inner.cache.invalidate_cache_state();
546
547        self.inner.try_general_repair().await?;
548        Ok(())
549    }
550
551    async fn find_start_address(&mut self) -> Result<NextAddress, Error<S::Error>> {
552        if self.inner.cache.is_dirty() {
553            self.inner.cache.invalidate_cache_state();
554        }
555
556        let oldest_page = self.find_oldest_page().await?;
557
558        // We start at the start of the oldest page
559        let current_address = match self.inner.cache.first_item_after_erased(oldest_page) {
560            Some(address) => address,
561            None => {
562                calculate_page_address::<S>(self.inner.flash_range.clone(), oldest_page)
563                    + S::WORD_SIZE as u32
564            }
565        };
566
567        Ok(NextAddress::Address(current_address))
568    }
569
570    /// Resets the flash in the entire given flash range.
571    ///
572    /// This is just a thin helper function as it just calls the flash's erase function.
573    pub fn erase_all(&mut self) -> impl Future<Output = Result<(), Error<S::Error>>> {
574        self.inner.erase_all()
575    }
576
577    /// Get the minimal overhead size per stored item for the given flash type.
578    ///
579    /// The associated data of each item is additionally padded to a full flash word size, but that's not part of this number.\
580    /// This means the full item length is `returned number + (data length).next_multiple_of(S::WORD_SIZE)`.
581    #[must_use]
582    pub const fn item_overhead_size() -> u32 {
583        GenericStorage::<S, C, ()>::item_overhead_size()
584    }
585
586    /// Destroy the instance to get back the flash and the cache.
587    ///
588    /// The cache can be passed to a new storage instance, but only for the same flash region and if nothing has changed in flash.
589    pub fn destroy(self) -> (S, C) {
590        self.inner.destroy()
591    }
592
593    /// Get a reference to the flash. Mutating the memory is at your own risk.
594    pub const fn flash(&mut self) -> &mut S {
595        self.inner.flash()
596    }
597
598    /// Get the flash range being used
599    pub const fn flash_range(&self) -> Range<u32> {
600        self.inner.flash_range()
601    }
602
603    /// Get a reference to the cache
604    #[cfg(any(test, fuzzing))]
605    pub const fn cache(&mut self) -> &mut C {
606        &mut self.inner.cache
607    }
608
609    #[cfg(any(test, feature = "std", fuzzing))]
610    /// Print all items in flash to the returned string
611    ///
612    /// This is meant as a debugging utility. The string format is not stable.
613    pub fn print_items(&mut self) -> impl Future<Output = String> {
614        self.inner.print_items()
615    }
616}
617
618#[derive(PartialEq, Eq, Clone, Copy, Debug)]
619enum PreviousItemStates {
620    AllPopped,
621    AllButCurrentPopped,
622    Unpopped,
623}
624
625/// An iterator-like interface for peeking into data stored in flash with the option to pop it.
626pub struct QueueIterator<'s, S: NorFlash, C: CacheImpl<()>> {
627    storage: &'s mut QueueStorage<S, C>,
628    next_address: NextAddress,
629    previous_item_states: PreviousItemStates,
630    oldest_page: usize,
631}
632
633impl<S: NorFlash, C: CacheImpl<()>> Debug for QueueIterator<'_, S, C> {
634    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
635        f.debug_struct("QueueIterator")
636            .field("current_address", &self.next_address)
637            .finish_non_exhaustive()
638    }
639}
640
641#[derive(Debug, Clone, Copy)]
642enum NextAddress {
643    Address(u32),
644    PageAfter(usize),
645}
646
647impl<'s, S: NorFlash, C: CacheImpl<()>> QueueIterator<'s, S, C> {
648    async fn new(storage: &'s mut QueueStorage<S, C>) -> Result<Self, Error<S::Error>> {
649        let start_address = run_with_auto_repair!(
650            function = storage.find_start_address().await,
651            repair = storage.try_repair().await?
652        )?;
653
654        let oldest_page = match start_address {
655            NextAddress::Address(address) => {
656                calculate_page_index::<S>(storage.inner.flash_range.clone(), address)
657            }
658            NextAddress::PageAfter(index) => storage.inner.next_page(index),
659        };
660
661        Ok(Self {
662            storage,
663            next_address: start_address,
664            previous_item_states: PreviousItemStates::AllPopped,
665            oldest_page,
666        })
667    }
668
669    /// Get the next entry.
670    ///
671    /// If there are no more entries, None is returned.
672    ///
673    /// The `data_buffer` has to be large enough to be able to hold the largest item in flash.
674    pub async fn next<'d, 'q>(
675        &'q mut self,
676        data_buffer: &'d mut [u8],
677    ) -> Result<Option<QueueIteratorEntry<'s, 'd, 'q, S, C>>, Error<S::Error>> {
678        // We continue from a place where the current item wasn't popped
679        // That means that from now on, the next item will have unpopped items behind it
680        if self.previous_item_states == PreviousItemStates::AllButCurrentPopped {
681            self.previous_item_states = PreviousItemStates::Unpopped;
682        }
683
684        let value = run_with_auto_repair!(
685            function = self.next_inner(data_buffer).await,
686            repair = self.storage.try_repair().await?
687        );
688
689        match value {
690            Ok(Some((item, address))) => Ok(Some(QueueIteratorEntry {
691                iter: self,
692                item: item.reborrow(data_buffer).ok_or_else(|| Error::LogicBug {
693                    #[cfg(feature = "_test")]
694                    backtrace: std::backtrace::Backtrace::capture(),
695                })?,
696                address,
697            })),
698            Ok(None) => Ok(None),
699            Err(e) => Err(e),
700        }
701    }
702
703    async fn next_inner(
704        &mut self,
705        data_buffer: &mut [u8],
706    ) -> Result<Option<(ItemUnborrowed, u32)>, Error<S::Error>> {
707        if self.storage.inner.cache.is_dirty() {
708            self.storage.inner.cache.invalidate_cache_state();
709        }
710
711        loop {
712            // Get the current page and address based on what was stored
713            let (current_page, current_address) = match self.next_address {
714                NextAddress::PageAfter(previous_page) => {
715                    let next_page = self.storage.inner.next_page(previous_page);
716                    if self
717                        .storage
718                        .inner
719                        .get_page_state_cached(next_page)
720                        .await?
721                        .is_open()
722                        || next_page == self.oldest_page
723                    {
724                        self.storage.inner.cache.unmark_dirty();
725                        return Ok(None);
726                    }
727
728                    // We now know the previous page was left because there were no items on there anymore
729                    // If we know all those items were popped, we can proactively open the previous page
730                    // This is amazing for performance
731                    if self.previous_item_states == PreviousItemStates::AllPopped {
732                        self.storage.inner.open_page(previous_page).await?;
733                    }
734
735                    let current_address = calculate_page_address::<S>(
736                        self.storage.inner.flash_range.clone(),
737                        next_page,
738                    ) + S::WORD_SIZE as u32;
739
740                    self.next_address = NextAddress::Address(current_address);
741
742                    (next_page, current_address)
743                }
744                NextAddress::Address(address) => (
745                    calculate_page_index::<S>(self.storage.inner.flash_range.clone(), address),
746                    address,
747                ),
748            };
749
750            let page_data_end_address = calculate_page_end_address::<S>(
751                self.storage.inner.flash_range.clone(),
752                current_page,
753            ) - S::WORD_SIZE as u32;
754
755            // Search for the first item with data
756            let mut it = ItemHeaderIter::new(current_address, page_data_end_address);
757            // No need to worry about cache here since that has been dealt with at the creation of this iterator
758            if let (Some(found_item_header), found_item_address) = it
759                .traverse(&mut self.storage.inner.flash, |header, _| {
760                    header.crc.is_none()
761                })
762                .await?
763            {
764                let maybe_item = found_item_header
765                    .read_item(
766                        &mut self.storage.inner.flash,
767                        data_buffer,
768                        found_item_address,
769                        page_data_end_address,
770                    )
771                    .await?;
772
773                match maybe_item {
774                    item::MaybeItem::Corrupted(header, _) => {
775                        let next_address = header.next_item_address::<S>(found_item_address);
776                        self.next_address = if next_address >= page_data_end_address {
777                            NextAddress::PageAfter(current_page)
778                        } else {
779                            NextAddress::Address(next_address)
780                        };
781                    }
782                    item::MaybeItem::Erased(_, _) => {
783                        // Item is already erased
784                        return Err(Error::LogicBug {
785                            #[cfg(feature = "_test")]
786                            backtrace: std::backtrace::Backtrace::capture(),
787                        });
788                    }
789                    item::MaybeItem::Present(item) => {
790                        let next_address = item.header.next_item_address::<S>(found_item_address);
791                        self.next_address = if next_address >= page_data_end_address {
792                            NextAddress::PageAfter(current_page)
793                        } else {
794                            NextAddress::Address(next_address)
795                        };
796
797                        // Record that the current item hasn't been popped (yet)
798                        if self.previous_item_states == PreviousItemStates::AllPopped {
799                            self.previous_item_states = PreviousItemStates::AllButCurrentPopped;
800                        }
801
802                        // Return the item we found
803                        self.storage.inner.cache.unmark_dirty();
804                        return Ok(Some((item.unborrow(), found_item_address)));
805                    }
806                }
807            } else {
808                self.next_address = NextAddress::PageAfter(current_page);
809            }
810        }
811    }
812}
813
814/// An entry in the iteration over the queue flash
815pub struct QueueIteratorEntry<'s, 'd, 'q, S: NorFlash, CI: CacheImpl<()>> {
816    iter: &'q mut QueueIterator<'s, S, CI>,
817    address: u32,
818    item: Item<'d>,
819}
820
821impl<S: NorFlash, CI: CacheImpl<()>> Deref for QueueIteratorEntry<'_, '_, '_, S, CI> {
822    type Target = [u8];
823
824    fn deref(&self) -> &Self::Target {
825        self.item.data()
826    }
827}
828
829impl<S: NorFlash, CI: CacheImpl<()>> DerefMut for QueueIteratorEntry<'_, '_, '_, S, CI> {
830    fn deref_mut(&mut self) -> &mut Self::Target {
831        self.item.data_mut()
832    }
833}
834
835impl<'d, S: NorFlash, CI: CacheImpl<()>> QueueIteratorEntry<'_, 'd, '_, S, CI> {
836    /// Get a mutable reference to the data of this entry, but consume the entry too.
837    /// This function has some relaxed lifetime constraints compared to the deref impls.
838    #[must_use]
839    pub fn into_buf(self) -> &'d mut [u8] {
840        self.item.data_owned()
841    }
842
843    /// Pop the data in flash that corresponds to this entry. This makes it so
844    /// future peeks won't find this data anymore.
845    pub async fn pop(self) -> Result<&'d mut [u8], Error<S::Error>>
846    where
847        S: MultiwriteNorFlash,
848    {
849        let (header, item_data_buffer) = self.item.header_and_data_owned();
850
851        // We're popping ourself, so if all previous but us were popped, then now all are popped again
852        if self.iter.previous_item_states == PreviousItemStates::AllButCurrentPopped {
853            self.iter.previous_item_states = PreviousItemStates::AllPopped;
854        }
855
856        header
857            .erase_data(
858                &mut self.iter.storage.inner.flash,
859                self.iter.storage.inner.flash_range.clone(),
860                &mut self.iter.storage.inner.cache,
861                self.address,
862            )
863            .await?;
864
865        self.iter.storage.inner.cache.unmark_dirty();
866        Ok(item_data_buffer)
867    }
868
869    /// Get the flash address of the item
870    #[cfg(feature = "_test")]
871    pub fn address(&self) -> u32 {
872        self.address
873    }
874}
875
876#[cfg(test)]
877mod tests {
878    use crate::{
879        AlignedBuf,
880        cache::Cache,
881        mock_flash::{self, FlashAverageStatsResult, FlashStatsResult, WriteCountCheck},
882    };
883
884    use super::*;
885    use futures_test::test;
886
887    type MockFlashBig = mock_flash::MockFlashBase<4, 4, 256>;
888    type MockFlashTiny = mock_flash::MockFlashBase<2, 1, 32>;
889
890    #[test]
891    async fn peek_and_overwrite_old_data() {
892        let mut storage = QueueStorage::new(
893            MockFlashTiny::new(WriteCountCheck::Twice, None, true),
894            const { QueueConfig::new(0x00..0x40) },
895            Cache::new_uncached(),
896        );
897        let mut data_buffer = AlignedBuf([0; 1024]);
898        const DATA_SIZE: usize = 22;
899
900        assert_eq!(storage.space_left().await.unwrap(), 60);
901
902        assert_eq!(storage.peek(&mut data_buffer).await.unwrap(), None);
903
904        data_buffer[..DATA_SIZE].copy_from_slice(&[0xAA; DATA_SIZE]);
905        storage
906            .push(&data_buffer[..DATA_SIZE], false)
907            .await
908            .unwrap();
909
910        assert_eq!(storage.space_left().await.unwrap(), 30);
911
912        assert_eq!(
913            storage.peek(&mut data_buffer).await.unwrap().unwrap(),
914            &[0xAA; DATA_SIZE]
915        );
916        data_buffer[..DATA_SIZE].copy_from_slice(&[0xBB; DATA_SIZE]);
917        storage
918            .push(&data_buffer[..DATA_SIZE], false)
919            .await
920            .unwrap();
921
922        assert_eq!(storage.space_left().await.unwrap(), 0);
923
924        assert_eq!(
925            storage.peek(&mut data_buffer).await.unwrap().unwrap(),
926            &[0xAA; DATA_SIZE]
927        );
928
929        // Flash is full, this should fail
930        data_buffer[..DATA_SIZE].copy_from_slice(&[0xCC; DATA_SIZE]);
931        storage
932            .push(&data_buffer[..DATA_SIZE], false)
933            .await
934            .unwrap_err();
935        // Now we allow overwrite, so it should work
936        data_buffer[..DATA_SIZE].copy_from_slice(&[0xDD; DATA_SIZE]);
937        storage.push(&data_buffer[..DATA_SIZE], true).await.unwrap();
938
939        assert_eq!(
940            storage.peek(&mut data_buffer).await.unwrap().unwrap(),
941            &[0xBB; DATA_SIZE]
942        );
943        assert_eq!(
944            storage.pop(&mut data_buffer).await.unwrap().unwrap(),
945            &[0xBB; DATA_SIZE]
946        );
947
948        assert_eq!(storage.space_left().await.unwrap(), 30);
949
950        assert_eq!(
951            storage.peek(&mut data_buffer).await.unwrap().unwrap(),
952            &[0xDD; DATA_SIZE]
953        );
954        assert_eq!(
955            storage.pop(&mut data_buffer).await.unwrap().unwrap(),
956            &[0xDD; DATA_SIZE]
957        );
958
959        assert_eq!(storage.space_left().await.unwrap(), 60);
960
961        assert_eq!(storage.peek(&mut data_buffer).await.unwrap(), None);
962        assert_eq!(storage.pop(&mut data_buffer).await.unwrap(), None);
963    }
964
965    #[test]
966    async fn push_pop() {
967        let mut storage = QueueStorage::new(
968            MockFlashBig::new(WriteCountCheck::Twice, None, true),
969            const { QueueConfig::new(0x000..0x1000) },
970            Cache::new_uncached(),
971        );
972
973        let mut data_buffer = AlignedBuf([0; 1024]);
974
975        for i in 0..2000 {
976            println!("{i}");
977            let data = vec![i as u8; i % 512 + 1];
978
979            storage.push(&data, true).await.unwrap();
980            assert_eq!(
981                storage.peek(&mut data_buffer).await.unwrap().unwrap(),
982                &data,
983                "At {i}"
984            );
985            assert_eq!(
986                storage.pop(&mut data_buffer).await.unwrap().unwrap(),
987                &data,
988                "At {i}"
989            );
990            assert_eq!(
991                storage.peek(&mut data_buffer).await.unwrap(),
992                None,
993                "At {i}"
994            );
995        }
996    }
997
998    #[test]
999    async fn iter_pop_out_of_order() {
1000        let mut storage = QueueStorage::new(
1001            MockFlashBig::new(WriteCountCheck::Twice, None, true),
1002            const { QueueConfig::new(0x000..0x1000) },
1003            Cache::new_uncached(),
1004        );
1005
1006        let mut data_buffer = AlignedBuf([0; 1024]);
1007
1008        let gen_data = |i: usize| vec![i as u8; i % 512 + 1];
1009        const COUNT: usize = 20;
1010
1011        for i in 0..COUNT {
1012            storage.push(&gen_data(i), false).await.unwrap();
1013        }
1014
1015        let mut iterator = storage.iter().await.unwrap();
1016        let mut i = 0;
1017        while let Some(entry) = iterator.next(&mut data_buffer).await.unwrap() {
1018            if i % 2 == 1 {
1019                assert_eq!(entry.pop().await.unwrap(), gen_data(i));
1020            }
1021
1022            i += 1;
1023        }
1024        assert_eq!(i, COUNT);
1025
1026        let mut iterator = storage.iter().await.unwrap();
1027        let mut i = 0;
1028        while let Some(entry) = iterator.next(&mut data_buffer).await.unwrap() {
1029            assert_eq!(entry.into_buf(), gen_data(i));
1030            i += 2;
1031        }
1032        assert_eq!(i, COUNT);
1033    }
1034
1035    #[test]
1036    async fn push_pop_tiny() {
1037        let mut storage = QueueStorage::new(
1038            MockFlashTiny::new(WriteCountCheck::Twice, None, true),
1039            const { QueueConfig::new(0x00..0x40) },
1040            Cache::new_uncached(),
1041        );
1042        let mut data_buffer = AlignedBuf([0; 1024]);
1043
1044        for i in 0..2000 {
1045            println!("{i}");
1046            let data = vec![i as u8; i % 20 + 1];
1047
1048            println!("PUSH");
1049            storage.push(&data, true).await.unwrap();
1050            assert_eq!(
1051                storage.peek(&mut data_buffer).await.unwrap().unwrap(),
1052                &data,
1053                "At {i}"
1054            );
1055            println!("POP");
1056            assert_eq!(
1057                storage.pop(&mut data_buffer).await.unwrap().unwrap(),
1058                &data,
1059                "At {i}"
1060            );
1061            println!("PEEK");
1062            assert_eq!(
1063                storage.peek(&mut data_buffer).await.unwrap(),
1064                None,
1065                "At {i}"
1066            );
1067            println!("DONE");
1068        }
1069    }
1070
1071    #[test]
1072    /// Same as [push_lots_then_pop_lots], except with added peeking and using the iterator style
1073    async fn push_peek_pop_many() {
1074        let mut storage = QueueStorage::new(
1075            MockFlashBig::new(WriteCountCheck::Twice, None, true),
1076            const { QueueConfig::new(0x000..0x1000) },
1077            Cache::new_uncached(),
1078        );
1079        let mut data_buffer = AlignedBuf([0; 1024]);
1080
1081        let mut push_stats = FlashStatsResult::default();
1082        let mut pushes = 0;
1083        let mut peek_stats = FlashStatsResult::default();
1084        let mut peeks = 0;
1085        let mut pop_stats = FlashStatsResult::default();
1086        let mut pops = 0;
1087
1088        for loop_index in 0..100 {
1089            println!("Loop index: {loop_index}");
1090
1091            for i in 0..20 {
1092                let start_snapshot = storage.flash().stats_snapshot();
1093                let data = vec![i as u8; 50];
1094                storage.push(&data, false).await.unwrap();
1095                pushes += 1;
1096                push_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1097            }
1098
1099            let start_snapshot = storage.flash().stats_snapshot();
1100            let mut iterator = storage.iter().await.unwrap();
1101            peek_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1102            for i in 0..5 {
1103                let start_snapshot = iterator.storage.flash().stats_snapshot();
1104                let data = [i as u8; 50];
1105                assert_eq!(
1106                    iterator
1107                        .next(&mut data_buffer)
1108                        .await
1109                        .unwrap()
1110                        .unwrap()
1111                        .deref(),
1112                    &data[..],
1113                    "At {i}"
1114                );
1115                peeks += 1;
1116                peek_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1117            }
1118
1119            let start_snapshot = storage.flash().stats_snapshot();
1120            let mut iterator = storage.iter().await.unwrap();
1121            pop_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1122            for i in 0..5 {
1123                let start_snapshot = iterator.storage.flash().stats_snapshot();
1124                let data = vec![i as u8; 50];
1125                assert_eq!(
1126                    iterator
1127                        .next(&mut data_buffer)
1128                        .await
1129                        .unwrap()
1130                        .unwrap()
1131                        .pop()
1132                        .await
1133                        .unwrap(),
1134                    &data,
1135                    "At {i}"
1136                );
1137                pops += 1;
1138                pop_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1139            }
1140
1141            for i in 20..25 {
1142                let start_snapshot = storage.flash().stats_snapshot();
1143                let data = vec![i as u8; 50];
1144                storage.push(&data, false).await.unwrap();
1145                pushes += 1;
1146                push_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1147            }
1148
1149            let start_snapshot = storage.flash().stats_snapshot();
1150            let mut iterator = storage.iter().await.unwrap();
1151            peek_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1152            for i in 5..25 {
1153                let start_snapshot = iterator.storage.flash().stats_snapshot();
1154                let data = vec![i as u8; 50];
1155                assert_eq!(
1156                    iterator
1157                        .next(&mut data_buffer)
1158                        .await
1159                        .unwrap()
1160                        .unwrap()
1161                        .deref(),
1162                    &data,
1163                    "At {i}"
1164                );
1165                peeks += 1;
1166                peek_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1167            }
1168
1169            let start_snapshot = storage.flash().stats_snapshot();
1170            let mut iterator = storage.iter().await.unwrap();
1171            pop_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1172            for i in 5..25 {
1173                let start_snapshot = iterator.storage.flash().stats_snapshot();
1174                let data = vec![i as u8; 50];
1175                assert_eq!(
1176                    iterator
1177                        .next(&mut data_buffer)
1178                        .await
1179                        .unwrap()
1180                        .unwrap()
1181                        .pop()
1182                        .await
1183                        .unwrap(),
1184                    &data,
1185                    "At {i}"
1186                );
1187                pops += 1;
1188                pop_stats += start_snapshot.compare_to(iterator.storage.flash().stats_snapshot());
1189            }
1190        }
1191
1192        // Assert the performance. These numbers can be changed if acceptable.
1193        approx::assert_relative_eq!(
1194            push_stats.take_average(pushes),
1195            FlashAverageStatsResult {
1196                avg_erases: 0.0,
1197                avg_reads: 16.864,
1198                avg_writes: 3.1252,
1199                avg_bytes_read: 105.4112,
1200                avg_bytes_written: 60.5008
1201            }
1202        );
1203        approx::assert_relative_eq!(
1204            peek_stats.take_average(peeks),
1205            FlashAverageStatsResult {
1206                avg_erases: 0.0052,
1207                avg_reads: 3.8656,
1208                avg_writes: 0.0,
1209                avg_bytes_read: 70.4256,
1210                avg_bytes_written: 0.0
1211            }
1212        );
1213        approx::assert_relative_eq!(
1214            pop_stats.take_average(pops),
1215            FlashAverageStatsResult {
1216                avg_erases: 0.0572,
1217                avg_reads: 3.7772,
1218                avg_writes: 1.0,
1219                avg_bytes_read: 69.7184,
1220                avg_bytes_written: 8.0
1221            }
1222        );
1223    }
1224
1225    #[test]
1226    async fn push_lots_then_pop_lots() {
1227        let mut storage = QueueStorage::new(
1228            MockFlashBig::new(WriteCountCheck::Twice, None, true),
1229            const { QueueConfig::new(0x000..0x1000) },
1230            Cache::new_uncached(),
1231        );
1232        let mut data_buffer = AlignedBuf([0; 1024]);
1233
1234        let mut push_stats = FlashStatsResult::default();
1235        let mut pushes = 0;
1236        let mut pop_stats = FlashStatsResult::default();
1237        let mut pops = 0;
1238
1239        for loop_index in 0..100 {
1240            println!("Loop index: {loop_index}");
1241
1242            for i in 0..20 {
1243                let start_snapshot = storage.flash().stats_snapshot();
1244                let data = vec![i as u8; 50];
1245                storage.push(&data, false).await.unwrap();
1246                pushes += 1;
1247                push_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1248            }
1249
1250            for i in 0..5 {
1251                let start_snapshot = storage.flash().stats_snapshot();
1252                let data = vec![i as u8; 50];
1253                assert_eq!(
1254                    storage.pop(&mut data_buffer).await.unwrap().unwrap(),
1255                    &data,
1256                    "At {i}"
1257                );
1258                pops += 1;
1259                pop_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1260            }
1261
1262            for i in 20..25 {
1263                let start_snapshot = storage.flash().stats_snapshot();
1264                let data = vec![i as u8; 50];
1265                storage.push(&data, false).await.unwrap();
1266                pushes += 1;
1267                push_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1268            }
1269
1270            for i in 5..25 {
1271                let start_snapshot = storage.flash().stats_snapshot();
1272                let data = vec![i as u8; 50];
1273                assert_eq!(
1274                    storage.pop(&mut data_buffer).await.unwrap().unwrap(),
1275                    &data,
1276                    "At {i}"
1277                );
1278                pops += 1;
1279                pop_stats += start_snapshot.compare_to(storage.flash().stats_snapshot());
1280            }
1281        }
1282
1283        // Assert the performance. These numbers can be changed if acceptable.
1284        approx::assert_relative_eq!(
1285            push_stats.take_average(pushes),
1286            FlashAverageStatsResult {
1287                avg_erases: 0.0,
1288                avg_reads: 16.864,
1289                avg_writes: 3.1252,
1290                avg_bytes_read: 105.4112,
1291                avg_bytes_written: 60.5008
1292            }
1293        );
1294        approx::assert_relative_eq!(
1295            pop_stats.take_average(pops),
1296            FlashAverageStatsResult {
1297                avg_erases: 0.0624,
1298                avg_reads: 23.5768,
1299                avg_writes: 1.0,
1300                avg_bytes_read: 180.512,
1301                avg_bytes_written: 8.0
1302            }
1303        );
1304    }
1305
1306    #[test]
1307    async fn pop_with_empty_section() {
1308        let mut storage = QueueStorage::new(
1309            MockFlashTiny::new(WriteCountCheck::Twice, None, true),
1310            const { QueueConfig::new(0x00..0x40) },
1311            Cache::new_uncached(),
1312        );
1313        let mut data_buffer = AlignedBuf([0; 1024]);
1314
1315        data_buffer[..20].copy_from_slice(&[0xAA; 20]);
1316        storage.push(&data_buffer[0..20], false).await.unwrap();
1317        data_buffer[..20].copy_from_slice(&[0xBB; 20]);
1318        storage.push(&data_buffer[0..20], false).await.unwrap();
1319
1320        // There's now an unused gap at the end of the first page
1321
1322        assert_eq!(
1323            storage.pop(&mut data_buffer).await.unwrap().unwrap(),
1324            &[0xAA; 20]
1325        );
1326
1327        assert_eq!(
1328            storage.pop(&mut data_buffer).await.unwrap().unwrap(),
1329            &[0xBB; 20]
1330        );
1331    }
1332
1333    #[test]
1334    async fn search_pages() {
1335        let mut storage = QueueStorage::new(
1336            MockFlashBig::new(WriteCountCheck::Twice, None, true),
1337            const { QueueConfig::new(0x000..0x1000) },
1338            Cache::new_uncached(),
1339        );
1340
1341        storage.inner.close_page(0).await.unwrap();
1342        storage.inner.close_page(1).await.unwrap();
1343        storage.inner.partial_close_page(2).await.unwrap();
1344
1345        assert_eq!(storage.find_youngest_page().await.unwrap(), 2);
1346        assert_eq!(storage.find_oldest_page().await.unwrap(), 0);
1347    }
1348
1349    #[test]
1350    async fn store_too_big_item() {
1351        let mut storage = QueueStorage::new(
1352            MockFlashBig::new(WriteCountCheck::Twice, None, true),
1353            const { QueueConfig::new(0x000..0x1000) },
1354            Cache::new_uncached(),
1355        );
1356
1357        storage
1358            .push(&AlignedBuf([0; 1024 - 4 * 2 - 8]), false)
1359            .await
1360            .unwrap();
1361
1362        assert_eq!(
1363            storage
1364                .push(&AlignedBuf([0; 1024 - 4 * 2 - 8 + 1]), false,)
1365                .await,
1366            Err(Error::ItemTooBig)
1367        );
1368    }
1369
1370    #[test]
1371    async fn push_on_single_page() {
1372        let mut storage = QueueStorage::new(
1373            mock_flash::MockFlashBase::<1, 4, 256>::new(WriteCountCheck::Twice, None, true),
1374            const { QueueConfig::new(0x000..0x400) },
1375            Cache::new_uncached(),
1376        );
1377
1378        for _ in 0..100 {
1379            match storage.push(&[0, 1, 2, 3, 4], true).await {
1380                Ok(_) => {}
1381                Err(e) => {
1382                    println!("{}", storage.print_items().await);
1383                    panic!("{e}");
1384                }
1385            }
1386        }
1387    }
1388
1389    #[test]
1390    async fn push_pop_on_single_page_without_overwrite() {
1391        let mut storage = QueueStorage::new(
1392            mock_flash::MockFlashBase::<1, 4, 1024>::new(WriteCountCheck::Twice, None, true),
1393            QueueConfig::new(0x0000..0x1000),
1394            Cache::new_uncached(),
1395        );
1396        let mut buf = [0; 80];
1397
1398        for i in 0..2000 {
1399            match storage.push(&AlignedBuf([0xAA; 32]), false).await {
1400                Ok(_) => {}
1401                Err(e) => {
1402                    println!("{}", storage.print_items().await);
1403                    panic!("push #{i} failed on an empty queue: {e}");
1404                }
1405            }
1406            assert_eq!(
1407                storage.pop(&mut buf).await.unwrap(),
1408                Some(&mut [0xAA; 32][..])
1409            );
1410            assert_eq!(storage.peek(&mut buf).await.unwrap(), None);
1411        }
1412    }
1413}
sequential_storage/queue/mod.rs

sequential_storage/queue/
mod.rs
