linera_views/views/
bucket_queue_view.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::{vec_deque::IterMut, VecDeque};
5
6use allocative::Allocative;
7use linera_base::data_types::ArithmeticError;
8#[cfg(with_metrics)]
9use linera_base::prometheus_util::MeasureLatency as _;
10use serde::{de::DeserializeOwned, Deserialize, Serialize};
11
12use crate::{
13    batch::Batch,
14    common::{from_bytes_option, from_bytes_option_or_default, HasherOutput},
15    context::Context,
16    hashable_wrapper::WrappedHashableContainerView,
17    historical_hash_wrapper::HistoricallyHashableView,
18    store::ReadableKeyValueStore as _,
19    views::{ClonableView, HashableView, Hasher, View, ViewError, MIN_VIEW_TAG},
20};
21
22#[cfg(with_metrics)]
23pub(crate) mod metrics {
24    use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
25    use prometheus::HistogramVec;
26
27    linera_base::declare_metrics! {
28        /// The runtime of hash computation
29        pub static BUCKET_QUEUE_VIEW_HASH_RUNTIME: HistogramVec =
30            register_histogram_vec(
31                "bucket_queue_view_hash_runtime",
32                "BucketQueueView hash runtime",
33                &[],
34                exponential_bucket_latencies(5.0),
35            );
36    }
37}
38
39/// Key tags to create the sub-keys of a [`BucketQueueView`] on top of the base key.
40#[repr(u8)]
41enum KeyTag {
42    /// Key tag for the `BucketLayout` metadata.
43    Layout = MIN_VIEW_TAG,
44    /// Key tag for the front bucket.
45    Front,
46    /// Key tag for the content of middle buckets.
47    Middle,
48    /// Key tag for the back bucket.
49    Back,
50}
51
52/// The metadata of the view in storage.
53///
54/// This is O(1) in size regardless of the number of buckets. The invariant is that
55/// all middle buckets (i.e. neither front nor back) have exactly N elements. Only
56/// the front bucket (tracked by `front_position`) and back bucket can be partial.
57#[derive(Clone, Debug, Default, Serialize, Deserialize)]
58struct BucketLayout {
59    /// The position of the front value in the front bucket.
60    front_position: u32,
61    /// The total number of stored buckets.
62    num_buckets: u32,
63    /// The logical index of the front bucket. Middle bucket at position `p` (1-indexed
64    /// from front) has storage key `KeyTag::Middle + (first_index + p)`.
65    first_index: u32,
66}
67
68/// The position of the queue's front value within the stored buckets.
69#[derive(Copy, Clone, Debug, Allocative)]
70struct Cursor {
71    /// The offset of the current front bucket from the front bucket of the saved
72    /// layout (`0` = the saved `stored_front` bucket).
73    offset: usize,
74    /// The position of the value within that bucket.
75    position: usize,
76}
77
78/// A view that supports a FIFO queue for values of type `T`.
79/// The size `N` has to be chosen by taking into account the size of the type `T`
80/// and the basic size of a block. For example a total size of 100 bytes to 10 KB
81/// seems adequate.
82///
83/// Only the endpoints of the stored sequence are materialized: `stored_front` and
84/// `stored_back` are kept in memory, while the middle buckets live solely in storage
85/// (each holds exactly `N` elements, so they are tracked by `stored_num_buckets` alone
86/// and read back lazily). The sole exception is `current_middle`, which holds the bucket
87/// the cursor currently points into once `delete_front` has advanced past the front.
88//#[allocative(bound = "T: Allocative")]
89#[derive(Debug, Allocative)]
90#[allocative(bound = "C, T: Allocative, const N: usize")]
91pub struct BucketQueueView<C, T, const N: usize> {
92    /// The view context.
93    #[allocative(skip)]
94    context: C,
95    /// The newly inserted back values, not yet persisted.
96    new_back_values: VecDeque<T>,
97    /// Storage index of the front bucket (the bucket at cursor offset `0`).
98    stored_first_index: u32,
99    /// The total number of stored buckets (front + middles + back).
100    stored_num_buckets: u32,
101    /// The front bucket's data, fully materialized; empty iff `stored_num_buckets == 0`.
102    ///
103    /// This is the saved front bucket and the anchor that `rollback` restores to: it is
104    /// never mutated by `delete_front`, so `rollback` (which is synchronous and reads
105    /// neither storage nor the live cursor) can rebuild the cursor from it together with
106    /// `stored_front_position`. In-memory mutations such as `clear` must leave it intact.
107    stored_front: Vec<T>,
108    /// The back bucket's data, fully materialized; empty when `stored_num_buckets < 2`
109    /// (a single stored bucket is represented by `stored_front` alone).
110    stored_back: Vec<T>,
111    /// Position of the front value within `stored_front`, as of the last save.
112    stored_front_position: u32,
113    /// The live cursor over the stored buckets, or `None` when no stored value remains
114    /// (so the front, if any, is the first of `new_back_values`).
115    ///
116    /// `None` does not imply `stored_num_buckets == 0`: after enough `delete_front`s the
117    /// cursor walks off the end and becomes `None` while the saved layout stays untouched
118    /// until the next save.
119    cursor: Option<Cursor>,
120    /// The data of the middle bucket the cursor currently points into, materialized when
121    /// `delete_front` has advanced strictly inside the middles (`0 < cursor.offset <
122    /// stored_num_buckets - 1`). `None` otherwise.
123    current_middle: Option<Vec<T>>,
124    /// Whether the storage is to be deleted or not.
125    delete_storage_first: bool,
126}
127
128impl<C, T, const N: usize> View for BucketQueueView<C, T, N>
129where
130    C: Context,
131    T: Send + Sync + Clone + Serialize + DeserializeOwned,
132{
133    const NUM_INIT_KEYS: usize = 3;
134
135    type Context = C;
136
137    fn context(&self) -> C {
138        self.context.clone()
139    }
140
141    fn pre_load(context: &C) -> Result<Vec<Vec<u8>>, ViewError> {
142        let key1 = context.base_key().base_tag(KeyTag::Layout as u8);
143        let key2 = context.base_key().base_tag(KeyTag::Front as u8);
144        let key3 = context.base_key().base_tag(KeyTag::Back as u8);
145        Ok(vec![key1, key2, key3])
146    }
147
148    fn post_load(context: C, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
149        let value_layout = values.first().ok_or(ViewError::PostLoadValuesError)?;
150        let value_front = values.get(1).ok_or(ViewError::PostLoadValuesError)?;
151        let value_back = values.get(2).ok_or(ViewError::PostLoadValuesError)?;
152        let front = from_bytes_option::<Vec<T>>(value_front)?;
153        let back = from_bytes_option::<Vec<T>>(value_back)?;
154        let layout = from_bytes_option_or_default::<BucketLayout>(value_layout)?;
155        // The front bucket is present iff there is stored data. The middle buckets are
156        // read back lazily and the back bucket is materialized only when `num_buckets >= 2`.
157        let (stored_first_index, stored_num_buckets, stored_front, stored_back, cursor) =
158            match front {
159                Some(front) => {
160                    let back = if layout.num_buckets >= 2 {
161                        back.unwrap_or_default()
162                    } else {
163                        Vec::new()
164                    };
165                    let cursor = Cursor {
166                        offset: 0,
167                        position: layout.front_position as usize,
168                    };
169                    (
170                        layout.first_index,
171                        layout.num_buckets,
172                        front,
173                        back,
174                        Some(cursor),
175                    )
176                }
177                None => (0, 0, Vec::new(), Vec::new(), None),
178            };
179        Ok(Self {
180            context,
181            new_back_values: VecDeque::new(),
182            stored_first_index,
183            stored_num_buckets,
184            stored_front,
185            stored_back,
186            stored_front_position: layout.front_position,
187            cursor,
188            current_middle: None,
189            delete_storage_first: false,
190        })
191    }
192
193    fn rollback(&mut self) {
194        // The saved layout (`stored_front`, `stored_back`, `stored_first_index`,
195        // `stored_num_buckets`, `stored_front_position`) is never mutated outside of a
196        // save, so restoring the last-saved state only requires resetting the live cursor
197        // and dropping the pending back values.
198        self.delete_storage_first = false;
199        self.cursor = (self.stored_num_buckets > 0).then_some(Cursor {
200            offset: 0,
201            position: self.stored_front_position as usize,
202        });
203        self.current_middle = None;
204        self.new_back_values.clear();
205    }
206
207    async fn has_pending_changes(&self) -> bool {
208        if self.delete_storage_first {
209            return true;
210        }
211        if self.stored_num_buckets > 0 {
212            let Some(cursor) = self.cursor else {
213                return true;
214            };
215            if cursor.offset != 0 || cursor.position != self.stored_front_position as usize {
216                return true;
217            }
218        }
219        !self.new_back_values.is_empty()
220    }
221
222    fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
223        let plan = self.save_plan()?;
224        match plan.case {
225            SaveCase::Empty => {
226                if plan.has_storage {
227                    batch.delete_key_prefix(self.context.base_key().bytes.clone());
228                }
229                Ok(true)
230            }
231            SaveCase::MetadataOnly => {
232                batch.put_key_value(
233                    self.layout_key(),
234                    &BucketLayout {
235                        front_position: plan.cursor_position,
236                        num_buckets: 1,
237                        first_index: self.stored_first_index,
238                    },
239                )?;
240                Ok(false)
241            }
242            SaveCase::Rewrite => {
243                if plan.has_storage {
244                    batch.delete_key_prefix(self.context.base_key().bytes.clone());
245                }
246                let mut all_data = Vec::new();
247                if let Some(data) = self.current_front_data() {
248                    all_data.extend(data[plan.cursor_position as usize..].iter().cloned());
249                }
250                all_data.extend(self.new_back_values.iter().cloned());
251                if all_data.is_empty() {
252                    return Ok(true);
253                }
254                let first_index = 0;
255                let num_buckets = self.write_chunks(batch, &all_data, first_index)?;
256                batch.put_key_value(
257                    self.layout_key(),
258                    &BucketLayout {
259                        front_position: 0,
260                        num_buckets,
261                        first_index,
262                    },
263                )?;
264                Ok(false)
265            }
266            SaveCase::Patch {
267                new_first_index,
268                remaining_count,
269            } => {
270                // Delete the keys of the consumed middle buckets (relative offsets
271                // 1..remaining_offset). Offset 0 was the old front (its KeyTag::Front is
272                // overwritten below if the front moved). The back is preserved since
273                // remaining_count >= 2.
274                for offset in 1..plan.remaining_offset {
275                    let index = checked_bucket_index(self.stored_first_index, offset)?;
276                    batch.delete_key(self.get_middle_key(index)?);
277                }
278                // Promote the new front bucket if the cursor crossed buckets.
279                if plan.remaining_offset > 0 {
280                    let data = self
281                        .current_front_data()
282                        .expect("Patch implies a live cursor within the stored buckets");
283                    batch.put_key_value(self.front_key(), &data.to_vec())?;
284                    batch.delete_key(self.get_middle_key(new_first_index)?);
285                }
286
287                let num_buckets = if self.new_back_values.is_empty() {
288                    remaining_count
289                } else {
290                    // Merge old back + new values, re-chunk into full-N middles + new back.
291                    let mut merged = self.stored_back.clone();
292                    merged.extend(self.new_back_values.iter().cloned());
293                    let chunks = merged.chunks(N).collect::<Vec<_>>();
294                    let num_new_chunks =
295                        u32::try_from(chunks.len()).map_err(|_| ArithmeticError::Overflow)?;
296                    // The old back becomes the first re-chunked bucket; it sits just past the
297                    // surviving middles (`remaining_count - 1` buckets after the new front).
298                    let new_middle_start =
299                        checked_bucket_index(new_first_index, remaining_count as usize - 1)?;
300                    for (i, chunk) in chunks.iter().enumerate().take(chunks.len() - 1) {
301                        let key =
302                            self.get_middle_key(checked_bucket_index(new_middle_start, i)?)?;
303                        batch.put_key_value(key, &chunk.to_vec())?;
304                    }
305                    batch.put_key_value(self.back_key(), &chunks.last().unwrap().to_vec())?;
306                    (remaining_count - 1)
307                        .checked_add(num_new_chunks)
308                        .ok_or(ArithmeticError::Overflow)?
309                };
310
311                batch.put_key_value(
312                    self.layout_key(),
313                    &BucketLayout {
314                        front_position: plan.cursor_position,
315                        num_buckets,
316                        first_index: new_first_index,
317                    },
318                )?;
319                Ok(false)
320            }
321        }
322    }
323
324    fn post_save(&mut self) {
325        let plan = self.save_plan().expect("verified in pre_save");
326        self.delete_storage_first = false;
327        match plan.case {
328            SaveCase::Empty => {
329                self.stored_first_index = 0;
330                self.stored_num_buckets = 0;
331                self.stored_front = Vec::new();
332                self.stored_back = Vec::new();
333                self.cursor = None;
334                self.current_middle = None;
335                self.stored_front_position = 0;
336            }
337            SaveCase::MetadataOnly => {
338                // The single front bucket survives unchanged; only the cursor advanced.
339                self.cursor = Some(Cursor {
340                    offset: 0,
341                    position: plan.cursor_position as usize,
342                });
343                self.current_middle = None;
344                self.stored_front_position = plan.cursor_position;
345            }
346            SaveCase::Rewrite => {
347                let mut all_data = Vec::new();
348                if let Some(data) = self.current_front_data() {
349                    all_data.extend(data[plan.cursor_position as usize..].iter().cloned());
350                }
351                all_data.extend(std::mem::take(&mut self.new_back_values));
352                // Mirror the `post_load` shape: only the front and back are materialized;
353                // the middles were written by `pre_save` and are read back lazily.
354                let num_chunks = all_data.chunks(N).len();
355                self.stored_first_index = 0;
356                self.stored_num_buckets = u32::try_from(num_chunks).expect("verified in pre_save");
357                self.current_middle = None;
358                self.stored_front_position = 0;
359                if num_chunks == 0 {
360                    self.stored_front = Vec::new();
361                    self.stored_back = Vec::new();
362                    self.cursor = None;
363                } else {
364                    self.stored_back = if num_chunks >= 2 {
365                        all_data[(num_chunks - 1) * N..].to_vec()
366                    } else {
367                        Vec::new()
368                    };
369                    all_data.truncate(N); // keep only the front chunk
370                    self.stored_front = all_data;
371                    self.cursor = Some(Cursor {
372                        offset: 0,
373                        position: 0,
374                    });
375                }
376            }
377            SaveCase::Patch {
378                new_first_index,
379                remaining_count,
380            } => {
381                let cursor = self.cursor.expect("Patch implies a live cursor");
382                // Promote the current front bucket to the new saved front. With >= 2
383                // buckets surviving the cursor is never on the back, so a moved front is
384                // always the materialized `current_middle`.
385                if plan.remaining_offset > 0 {
386                    self.stored_front = self
387                        .current_middle
388                        .take()
389                        .expect("the middle the cursor points into is loaded");
390                }
391                self.current_middle = None;
392                self.stored_first_index = new_first_index;
393                self.stored_num_buckets = if self.new_back_values.is_empty() {
394                    remaining_count
395                } else {
396                    // Merge old back + new values; the last chunk is the new back and the
397                    // earlier chunks are new middles (storage-only).
398                    let mut merged = std::mem::take(&mut self.stored_back);
399                    merged.extend(std::mem::take(&mut self.new_back_values));
400                    let num_new_chunks =
401                        u32::try_from(merged.chunks(N).len()).expect("verified in pre_save");
402                    let back_start = (num_new_chunks as usize - 1) * N;
403                    self.stored_back = merged.split_off(back_start);
404                    (remaining_count - 1) + num_new_chunks
405                };
406                self.cursor = Some(Cursor {
407                    offset: 0,
408                    position: cursor.position,
409                });
410                self.stored_front_position = plan.cursor_position;
411            }
412        }
413    }
414
415    fn clear(&mut self) {
416        // Leaves the saved layout in place (the `rollback` anchor); the next save sees
417        // `delete_storage_first` and wipes storage regardless.
418        self.delete_storage_first = true;
419        self.new_back_values.clear();
420        self.cursor = None;
421        self.current_middle = None;
422    }
423}
424
425impl<C: Clone, T: Clone, const N: usize> ClonableView for BucketQueueView<C, T, N>
426where
427    Self: View,
428{
429    fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
430        Ok(BucketQueueView {
431            context: self.context.clone(),
432            new_back_values: self.new_back_values.clone(),
433            stored_first_index: self.stored_first_index,
434            stored_num_buckets: self.stored_num_buckets,
435            stored_front: self.stored_front.clone(),
436            stored_back: self.stored_back.clone(),
437            stored_front_position: self.stored_front_position,
438            cursor: self.cursor,
439            current_middle: self.current_middle.clone(),
440            delete_storage_first: self.delete_storage_first,
441        })
442    }
443}
444
445/// Pattern describing how a save will affect storage and in-memory state.
446/// Derived from `&self` in [`BucketQueueView::save_plan`] and consumed by both
447/// `pre_save` (storage batch) and `post_save` (in-memory state).
448#[derive(Debug)]
449enum SaveCase {
450    /// Drop everything — leaves the queue empty.
451    Empty,
452    /// The single stored bucket survives without structural change; only the
453    /// cursor advanced inside the front bucket. Just refresh `front_position`.
454    MetadataOnly,
455    /// At most one stored bucket survives (after consumption) and there may be
456    /// new values to append. Drop all keys and rewrite from scratch.
457    Rewrite,
458    /// Two or more stored buckets survive. Delete consumed middles, promote a
459    /// surviving middle to front if the cursor crossed buckets, and — if there
460    /// are new values — merge them into the back and re-chunk.
461    Patch {
462        new_first_index: u32,
463        remaining_count: u32,
464    },
465}
466
467#[derive(Debug)]
468struct SavePlan {
469    case: SaveCase,
470    /// Relative bucket offset of the first surviving bucket (`cursor.offset`, or
471    /// `stored_num_buckets` when everything is dropped).
472    remaining_offset: usize,
473    /// Position of the cursor within the front bucket, validated as `u32` (the type
474    /// stored in `BucketLayout`). Cast to `usize` at the few sites that index into a
475    /// bucket's data.
476    cursor_position: u32,
477    /// True if there is any existing storage to clear (for `Empty`/`Rewrite`).
478    has_storage: bool,
479}
480
481/// Adds a relative `offset` to a base bucket index with checked arithmetic.
482///
483/// Bucket indices are `u32` (the width stored in `BucketLayout`); a queue would need
484/// billions of bucket rotations to overflow, but the additions that walk a base index
485/// across bucket offsets are guarded here to stay consistent with the rest of the view.
486fn checked_bucket_index(base: u32, offset: usize) -> Result<u32, ArithmeticError> {
487    let offset = u32::try_from(offset).map_err(|_| ArithmeticError::Overflow)?;
488    base.checked_add(offset).ok_or(ArithmeticError::Overflow)
489}
490
491impl<C: Context, T, const N: usize> BucketQueueView<C, T, N> {
492    fn front_key(&self) -> Vec<u8> {
493        self.context.base_key().base_tag(KeyTag::Front as u8)
494    }
495
496    fn back_key(&self) -> Vec<u8> {
497        self.context.base_key().base_tag(KeyTag::Back as u8)
498    }
499
500    fn layout_key(&self) -> Vec<u8> {
501        self.context.base_key().base_tag(KeyTag::Layout as u8)
502    }
503
504    /// Gets the key for a middle bucket with the given storage index.
505    fn get_middle_key(&self, index: u32) -> Result<Vec<u8>, ViewError> {
506        Ok(self
507            .context
508            .base_key()
509            .derive_tag_key(KeyTag::Middle as u8, &index)?)
510    }
511
512    /// The data of the bucket the live cursor points into (the queue front within the
513    /// stored buckets), or `None` when the stored portion is exhausted.
514    fn current_front_data(&self) -> Option<&[T]> {
515        let cursor = self.cursor?;
516        let num_buckets = self.stored_num_buckets as usize;
517        Some(if cursor.offset == 0 {
518            &self.stored_front
519        } else if cursor.offset + 1 == num_buckets {
520            &self.stored_back
521        } else {
522            self.current_middle
523                .as_deref()
524                .expect("the middle bucket the cursor points into is loaded")
525        })
526    }
527
528    /// The number of elements in the stored bucket at relative `offset`. Middle buckets
529    /// always hold exactly `N` (invariant), so they need not be materialized.
530    fn bucket_len(&self, offset: usize) -> usize {
531        if offset == 0 {
532            self.stored_front.len()
533        } else if offset + 1 == self.stored_num_buckets as usize {
534            self.stored_back.len()
535        } else {
536            N
537        }
538    }
539
540    /// Classifies the pending save based on the current view state.
541    /// Called once by `pre_save` and once by `post_save`; since `&self` is
542    /// unchanged between the two it returns the same plan both times.
543    fn save_plan(&self) -> Result<SavePlan, ViewError> {
544        let num_buckets = self.stored_num_buckets as usize;
545        let remaining_offset = if self.delete_storage_first {
546            num_buckets
547        } else {
548            self.cursor.map_or(num_buckets, |c| c.offset)
549        };
550        let remaining_count = num_buckets - remaining_offset;
551        let cursor_position = u32::try_from(self.cursor.map_or(0, |c| c.position))
552            .map_err(|_| ArithmeticError::Overflow)?;
553        let has_storage = self.stored_num_buckets > 0 || self.delete_storage_first;
554        let new_back_empty = self.new_back_values.is_empty();
555
556        let case = if remaining_count == 0 && new_back_empty {
557            SaveCase::Empty
558        } else if remaining_count == 1 && remaining_offset == 0 && new_back_empty {
559            SaveCase::MetadataOnly
560        } else if remaining_count <= 1 {
561            SaveCase::Rewrite
562        } else {
563            SaveCase::Patch {
564                new_first_index: checked_bucket_index(self.stored_first_index, remaining_offset)?,
565                remaining_count: u32::try_from(remaining_count)
566                    .map_err(|_| ArithmeticError::Overflow)?,
567            }
568        };
569        Ok(SavePlan {
570            case,
571            remaining_offset,
572            cursor_position,
573            has_storage,
574        })
575    }
576
577    /// Splits `data` into N-sized chunks and writes them as front (KeyTag::Front),
578    /// middles (KeyTag::Middle, starting at `first_index`+1), and back (KeyTag::Back).
579    /// Returns the total number of buckets written. The caller is responsible for
580    /// writing the matching `BucketLayout` entry. Used by `Rewrite`.
581    fn write_chunks(
582        &self,
583        batch: &mut Batch,
584        data: &[T],
585        first_index: u32,
586    ) -> Result<u32, ViewError>
587    where
588        T: Serialize + Clone,
589    {
590        let chunks = data.chunks(N).collect::<Vec<_>>();
591        let num_buckets = u32::try_from(chunks.len()).map_err(|_| ArithmeticError::Overflow)?;
592        batch.put_key_value(self.front_key(), &chunks[0].to_vec())?;
593        for (i, chunk) in chunks
594            .iter()
595            .enumerate()
596            .skip(1)
597            .take(chunks.len().saturating_sub(2))
598        {
599            let key = self.get_middle_key(checked_bucket_index(first_index, i)?)?;
600            batch.put_key_value(key, &chunk.to_vec())?;
601        }
602        if num_buckets >= 2 {
603            batch.put_key_value(self.back_key(), &chunks.last().unwrap().to_vec())?;
604        }
605        Ok(num_buckets)
606    }
607
608    /// Gets the number of entries that are in the container and in storage.
609    fn stored_count(&self) -> usize {
610        if self.delete_storage_first {
611            return 0;
612        }
613        let Some(cursor) = self.cursor else {
614            return 0;
615        };
616        let remaining = self.stored_num_buckets as usize - cursor.offset;
617        // Current front bucket: count the valid elements after the cursor position.
618        let front_count = self.bucket_len(cursor.offset) - cursor.position;
619        if remaining == 1 {
620            return front_count;
621        }
622        // Back bucket plus the full middles between the cursor and the back.
623        let back_count = self.stored_back.len();
624        let num_middles = remaining - 2;
625        front_count + num_middles * N + back_count
626    }
627
628    /// The total number of entries of the container.
629    /// ```rust
630    /// # tokio_test::block_on(async {
631    /// # use linera_views::context::MemoryContext;
632    /// # use linera_views::bucket_queue_view::BucketQueueView;
633    /// # use crate::linera_views::views::View;
634    /// # let context = MemoryContext::new_for_testing(());
635    /// let mut queue = BucketQueueView::<_, u8, 5>::load(context).await.unwrap();
636    /// queue.push_back(34);
637    /// assert_eq!(queue.count(), 1);
638    /// # })
639    /// ```
640    pub fn count(&self) -> usize {
641        self.stored_count() + self.new_back_values.len()
642    }
643}
644
645impl<C: Context, T: DeserializeOwned + Clone, const N: usize> BucketQueueView<C, T, N> {
646    /// Gets a reference on the front value if any.
647    /// ```rust
648    /// # tokio_test::block_on(async {
649    /// # use linera_views::context::MemoryContext;
650    /// # use linera_views::bucket_queue_view::BucketQueueView;
651    /// # use crate::linera_views::views::View;
652    /// # let context = MemoryContext::new_for_testing(());
653    /// let mut queue = BucketQueueView::<_, u8, 5>::load(context).await.unwrap();
654    /// queue.push_back(34);
655    /// queue.push_back(42);
656    /// assert_eq!(queue.front().cloned(), Some(34));
657    /// # })
658    /// ```
659    pub fn front(&self) -> Option<&T> {
660        match self.cursor {
661            Some(cursor) => {
662                let data = self
663                    .current_front_data()
664                    .expect("cursor is Some, so the current front is available");
665                Some(&data[cursor.position])
666            }
667            None => self.new_back_values.front(),
668        }
669    }
670
671    /// Reads the front value, if any.
672    /// ```rust
673    /// # tokio_test::block_on(async {
674    /// # use linera_views::context::MemoryContext;
675    /// # use linera_views::bucket_queue_view::BucketQueueView;
676    /// # use crate::linera_views::views::View;
677    /// # let context = MemoryContext::new_for_testing(());
678    /// let mut queue = BucketQueueView::<_, u8, 5>::load(context).await.unwrap();
679    /// queue.push_back(34);
680    /// queue.push_back(42);
681    /// let front = queue.front_mut().unwrap();
682    /// *front = 43;
683    /// assert_eq!(queue.front().cloned(), Some(43));
684    /// # })
685    /// ```
686    pub fn front_mut(&mut self) -> Option<&mut T> {
687        match self.cursor {
688            Some(cursor) => {
689                let num_buckets = self.stored_num_buckets as usize;
690                let data = if cursor.offset == 0 {
691                    &mut self.stored_front
692                } else if cursor.offset + 1 == num_buckets {
693                    &mut self.stored_back
694                } else {
695                    self.current_middle
696                        .as_mut()
697                        .expect("the middle bucket the cursor points into is loaded")
698                };
699                Some(
700                    data.get_mut(cursor.position)
701                        .expect("cursor.position must be a valid position within the front bucket"),
702                )
703            }
704            None => self.new_back_values.front_mut(),
705        }
706    }
707
708    /// Deletes the front value, if any.
709    /// ```rust
710    /// # tokio_test::block_on(async {
711    /// # use linera_views::context::MemoryContext;
712    /// # use linera_views::bucket_queue_view::BucketQueueView;
713    /// # use crate::linera_views::views::View;
714    /// # let context = MemoryContext::new_for_testing(());
715    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
716    /// queue.push_back(34 as u128);
717    /// queue.delete_front().await.unwrap();
718    /// assert_eq!(queue.elements().await.unwrap(), Vec::<u128>::new());
719    /// # })
720    /// ```
721    pub async fn delete_front(&mut self) -> Result<(), ViewError> {
722        let Some(cursor) = self.cursor else {
723            self.new_back_values.pop_front();
724            return Ok(());
725        };
726        let current_len = self
727            .current_front_data()
728            .expect("cursor points into the stored buckets")
729            .len();
730        let num_buckets = self.stored_num_buckets as usize;
731        let mut offset = cursor.offset;
732        let mut position = cursor.position + 1;
733        if position == current_len {
734            offset += 1;
735            position = 0;
736            if offset == num_buckets {
737                // The stored portion is now exhausted.
738                self.cursor = None;
739                self.current_middle = None;
740                return Ok(());
741            }
742            // The cursor crossed into the bucket at `offset` (>= 1). Materialize it as the
743            // new front *before* moving the cursor, so a failed load leaves the view's
744            // invariant intact (the current front bucket is always materialized).
745            if offset + 1 == num_buckets {
746                // The back bucket is already materialized.
747                self.current_middle = None;
748            } else {
749                let index = checked_bucket_index(self.stored_first_index, offset)?;
750                let key = self.get_middle_key(index)?;
751                let data = self
752                    .context
753                    .store()
754                    .read_value(&key)
755                    .await?
756                    .ok_or_else(|| {
757                        ViewError::MissingEntries("BucketQueueView::delete_front".into())
758                    })?;
759                self.current_middle = Some(data);
760            }
761        }
762        self.cursor = Some(Cursor { offset, position });
763        Ok(())
764    }
765
766    /// Pushes a value to the end of the queue.
767    /// ```rust
768    /// # tokio_test::block_on(async {
769    /// # use linera_views::context::MemoryContext;
770    /// # use linera_views::bucket_queue_view::BucketQueueView;
771    /// # use crate::linera_views::views::View;
772    /// # let context = MemoryContext::new_for_testing(());
773    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
774    /// queue.push_back(34);
775    /// assert_eq!(queue.elements().await.unwrap(), vec![34]);
776    /// # })
777    /// ```
778    pub fn push_back(&mut self, value: T) {
779        self.new_back_values.push_back(value);
780    }
781
782    /// Returns the list of elements in the queue.
783    /// ```rust
784    /// # tokio_test::block_on(async {
785    /// # use linera_views::context::MemoryContext;
786    /// # use linera_views::bucket_queue_view::BucketQueueView;
787    /// # use crate::linera_views::views::View;
788    /// # let context = MemoryContext::new_for_testing(());
789    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
790    /// queue.push_back(34);
791    /// queue.push_back(37);
792    /// assert_eq!(queue.elements().await.unwrap(), vec![34, 37]);
793    /// # })
794    /// ```
795    pub async fn elements(&self) -> Result<Vec<T>, ViewError> {
796        let count = self.count();
797        self.read_context(self.cursor, count).await
798    }
799
800    /// Returns the last element of a bucket queue view
801    /// ```rust
802    /// # tokio_test::block_on(async {
803    /// # use linera_views::context::MemoryContext;
804    /// # use linera_views::bucket_queue_view::BucketQueueView;
805    /// # use crate::linera_views::views::View;
806    /// # let context = MemoryContext::new_for_testing(());
807    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
808    /// queue.push_back(34);
809    /// queue.push_back(37);
810    /// assert_eq!(queue.back().await.unwrap(), Some(37));
811    /// # })
812    /// ```
813    pub async fn back(&mut self) -> Result<Option<T>, ViewError>
814    where
815        T: Clone,
816    {
817        if let Some(value) = self.new_back_values.back() {
818            return Ok(Some(value.clone()));
819        }
820        if self.cursor.is_none() {
821            return Ok(None);
822        }
823        // The last stored element is the back bucket's last (or the front's, for a
824        // single stored bucket).
825        let last = if self.stored_num_buckets >= 2 {
826            self.stored_back.last()
827        } else {
828            self.stored_front.last()
829        };
830        Ok(last.cloned())
831    }
832
833    async fn read_context(
834        &self,
835        cursor: Option<Cursor>,
836        count: usize,
837    ) -> Result<Vec<T>, ViewError> {
838        if count == 0 {
839            return Ok(Vec::new());
840        }
841        let mut elements = Vec::<T>::new();
842        let mut count_remain = count;
843        if let Some(cursor) = cursor {
844            let num_buckets = self.stored_num_buckets as usize;
845            // First pass: gather the storage keys of the middle buckets we will read.
846            let mut keys = Vec::new();
847            let mut position = cursor.position;
848            let mut remain = count;
849            for offset in cursor.offset..num_buckets {
850                if offset != 0 && offset + 1 != num_buckets {
851                    let index = checked_bucket_index(self.stored_first_index, offset)?;
852                    keys.push(self.get_middle_key(index)?);
853                }
854                let size = self.bucket_len(offset) - position;
855                if size >= remain {
856                    break;
857                }
858                remain -= size;
859                position = 0;
860            }
861            let values = self.context.store().read_multi_values_bytes(&keys).await?;
862            // Second pass: assemble the elements, reading middles from `values`.
863            let mut value_pos = 0;
864            let mut position = cursor.position;
865            for offset in cursor.offset..num_buckets {
866                let read_buf;
867                let data: &[T] = if offset == 0 {
868                    &self.stored_front
869                } else if offset + 1 == num_buckets {
870                    &self.stored_back
871                } else {
872                    let value = values[value_pos].as_ref().ok_or_else(|| {
873                        ViewError::MissingEntries("BucketQueueView::read_context".into())
874                    })?;
875                    value_pos += 1;
876                    read_buf = bcs::from_bytes::<Vec<T>>(value)?;
877                    &read_buf
878                };
879                let size = data.len() - position;
880                elements.extend(data[position..].iter().take(count_remain).cloned());
881                if size >= count_remain {
882                    return Ok(elements);
883                }
884                count_remain -= size;
885                position = 0;
886            }
887        }
888        let count_read = std::cmp::min(count_remain, self.new_back_values.len());
889        elements.extend(self.new_back_values.range(0..count_read).cloned());
890        Ok(elements)
891    }
892
893    /// Returns the first elements of a bucket queue view
894    /// ```rust
895    /// # tokio_test::block_on(async {
896    /// # use linera_views::context::MemoryContext;
897    /// # use linera_views::bucket_queue_view::BucketQueueView;
898    /// # use crate::linera_views::views::View;
899    /// # let context = MemoryContext::new_for_testing(());
900    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
901    /// queue.push_back(34);
902    /// queue.push_back(37);
903    /// queue.push_back(47);
904    /// assert_eq!(queue.read_front(2).await.unwrap(), vec![34, 37]);
905    /// # })
906    /// ```
907    pub async fn read_front(&self, count: usize) -> Result<Vec<T>, ViewError> {
908        let count = std::cmp::min(count, self.count());
909        self.read_context(self.cursor, count).await
910    }
911
912    /// Returns the last element of a bucket queue view
913    /// ```rust
914    /// # tokio_test::block_on(async {
915    /// # use linera_views::context::MemoryContext;
916    /// # use linera_views::bucket_queue_view::BucketQueueView;
917    /// # use crate::linera_views::views::View;
918    /// # let context = MemoryContext::new_for_testing(());
919    /// let mut queue = BucketQueueView::<_, u128, 5>::load(context).await.unwrap();
920    /// queue.push_back(34);
921    /// queue.push_back(37);
922    /// queue.push_back(47);
923    /// assert_eq!(queue.read_back(2).await.unwrap(), vec![37, 47]);
924    /// # })
925    /// ```
926    pub async fn read_back(&self, count: usize) -> Result<Vec<T>, ViewError> {
927        let count = std::cmp::min(count, self.count());
928        if count <= self.new_back_values.len() {
929            let start = self.new_back_values.len() - count;
930            Ok(self
931                .new_back_values
932                .range(start..)
933                .cloned()
934                .collect::<Vec<_>>())
935        } else {
936            let mut increment = self.count() - count;
937            let Some(cursor) = self.cursor else {
938                unreachable!("Cursor should be Some when stored_count > 0");
939            };
940            let num_buckets = self.stored_num_buckets as usize;
941            let mut position = cursor.position;
942            for offset in cursor.offset..num_buckets {
943                let size = self.bucket_len(offset) - position;
944                if increment < size {
945                    return self
946                        .read_context(
947                            Some(Cursor {
948                                offset,
949                                position: position + increment,
950                            }),
951                            count,
952                        )
953                        .await;
954                }
955                increment -= size;
956                position = 0;
957            }
958            unreachable!("BucketQueueView::read_back: iterated past all stored buckets without finding the requested position");
959        }
960    }
961
962    async fn load_all(&mut self) -> Result<(), ViewError> {
963        if !self.delete_storage_first {
964            let elements = self.elements().await?;
965            self.new_back_values.clear();
966            for elt in elements {
967                self.new_back_values.push_back(elt);
968            }
969            self.cursor = None;
970            self.current_middle = None;
971            self.delete_storage_first = true;
972        }
973        Ok(())
974    }
975
976    /// Gets a mutable iterator on the entries of the queue
977    /// ```rust
978    /// # tokio_test::block_on(async {
979    /// # use linera_views::context::MemoryContext;
980    /// # use linera_views::bucket_queue_view::BucketQueueView;
981    /// # use linera_views::views::View;
982    /// # let context = MemoryContext::new_for_testing(());
983    /// let mut queue = BucketQueueView::<_, u8, 5>::load(context).await.unwrap();
984    /// queue.push_back(34);
985    /// let mut iter = queue.try_iter_mut().await.unwrap();
986    /// let value = iter.next().unwrap();
987    /// *value = 42;
988    /// assert_eq!(queue.elements().await.unwrap(), vec![42]);
989    /// # })
990    /// ```
991    pub async fn try_iter_mut(&mut self) -> Result<IterMut<'_, T>, ViewError> {
992        self.load_all().await?;
993        Ok(self.new_back_values.iter_mut())
994    }
995}
996
997impl<C: Context, T: Serialize + DeserializeOwned + Send + Sync + Clone, const N: usize> HashableView
998    for BucketQueueView<C, T, N>
999where
1000    Self: View,
1001{
1002    type Hasher = sha3::Sha3_256;
1003
1004    async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1005        self.hash().await
1006    }
1007
1008    async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1009        #[cfg(with_metrics)]
1010        let _hash_latency = metrics::BUCKET_QUEUE_VIEW_HASH_RUNTIME.measure_latency();
1011        let elements = self.elements().await?;
1012        let mut hasher = sha3::Sha3_256::default();
1013        hasher.update_with_bcs_bytes(&elements)?;
1014        Ok(hasher.finalize())
1015    }
1016}
1017
1018/// Type wrapping `QueueView` while memoizing the hash.
1019pub type HashedBucketQueueView<C, T, const N: usize> =
1020    WrappedHashableContainerView<C, BucketQueueView<C, T, N>, HasherOutput>;
1021
1022/// Wrapper around `BucketQueueView` to compute hashes based on the history of changes.
1023pub type HistoricallyHashedBucketQueueView<C, T, const N: usize> =
1024    HistoricallyHashableView<C, BucketQueueView<C, T, N>>;
1025
1026#[cfg(with_graphql)]
1027mod graphql {
1028    use std::borrow::Cow;
1029
1030    use linera_base::data_types::ArithmeticError;
1031
1032    use super::BucketQueueView;
1033    use crate::{
1034        context::Context,
1035        graphql::{hash_name, mangle},
1036    };
1037
1038    impl<C: Send + Sync, T: async_graphql::OutputType, const N: usize> async_graphql::TypeName
1039        for BucketQueueView<C, T, N>
1040    {
1041        fn type_name() -> Cow<'static, str> {
1042            format!(
1043                "BucketQueueView_{}_{:08x}",
1044                mangle(T::type_name()),
1045                hash_name::<T>()
1046            )
1047            .into()
1048        }
1049    }
1050
1051    #[async_graphql::Object(cache_control(no_cache), name_type)]
1052    impl<C: Context, T: async_graphql::OutputType, const N: usize> BucketQueueView<C, T, N>
1053    where
1054        C: Send + Sync,
1055        T: serde::ser::Serialize + serde::de::DeserializeOwned + Clone + Send + Sync,
1056    {
1057        #[graphql(derived(name = "count"))]
1058        async fn count_(&self) -> Result<u32, async_graphql::Error> {
1059            Ok(u32::try_from(self.count()).map_err(|_| ArithmeticError::Overflow)?)
1060        }
1061
1062        async fn entries(&self, count: Option<usize>) -> async_graphql::Result<Vec<T>> {
1063            Ok(self
1064                .read_front(count.unwrap_or_else(|| self.count()))
1065                .await?)
1066        }
1067    }
1068}
1069
1070#[cfg(test)]
1071mod tests {
1072    use super::*;
1073    use crate::{
1074        batch::Batch,
1075        context::{Context, MemoryContext},
1076        store::WritableKeyValueStore as _,
1077    };
1078
1079    /// Regression test: a failed load while advancing the cursor in `delete_front`
1080    /// must not leave the view in a state where the current front bucket is not
1081    /// materialized. The next bucket is loaded *before* the cursor advances, so a
1082    /// failed load leaves the cursor (and `current_middle`) untouched.
1083    #[tokio::test]
1084    async fn delete_front_load_failure_preserves_invariant() -> Result<(), ViewError> {
1085        let context = MemoryContext::new_for_testing(());
1086        let mut view = BucketQueueView::<_, u8, 2>::load(context.clone()).await?;
1087        // Six elements -> front=[1,2], middle=[3,4] at index 1, back=[5,6].
1088        for value in [1u8, 2, 3, 4, 5, 6] {
1089            view.push_back(value);
1090        }
1091        save(&context, &mut view).await?;
1092
1093        let mut view = BucketQueueView::<_, u8, 2>::load(context.clone()).await?;
1094
1095        // Delete the middle bucket so that loading it during `delete_front` fails.
1096        let middle_key = view.get_middle_key(1)?;
1097        let mut batch = Batch::new();
1098        batch.delete_key(middle_key);
1099        context.store().write_batch(batch).await?;
1100
1101        view.delete_front().await?;
1102        let err = view.delete_front().await.expect_err("load should fail");
1103        assert!(matches!(err, ViewError::MissingEntries(_)));
1104
1105        save(&context, &mut view).await?;
1106
1107        Ok(())
1108    }
1109
1110    /// Roundtrip a queue through save/reload at several sizes around `N` to
1111    /// exercise the front/middle/back layout: empty, partial-front-only,
1112    /// exactly-one-bucket, front+back without middles, and several layouts with
1113    /// middle buckets.
1114    #[tokio::test]
1115    async fn save_load_roundtrip_across_sizes() -> Result<(), ViewError> {
1116        const N: usize = 3;
1117        for size in [0usize, 1, 2, N, N + 1, 2 * N, 2 * N + 1, 5 * N, 5 * N - 1] {
1118            let context = MemoryContext::new_for_testing(());
1119            let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1120            for i in 0..u32::try_from(size).unwrap() {
1121                view.push_back(i);
1122            }
1123            save(&context, &mut view).await?;
1124
1125            let reloaded = BucketQueueView::<_, u32, N>::load(context).await?;
1126            let elements = reloaded.elements().await?;
1127            let expected = (0..u32::try_from(size).unwrap()).collect::<Vec<_>>();
1128            assert_eq!(elements, expected, "size = {size}");
1129            assert_eq!(reloaded.count(), size, "count for size = {size}");
1130        }
1131        Ok(())
1132    }
1133
1134    /// Middle buckets must always contain exactly `N` elements after save — this is
1135    /// the invariant that lets the layout track them by count alone. Read the middle
1136    /// bucket keys back from storage and check each holds `N` elements.
1137    #[tokio::test]
1138    async fn middle_buckets_are_always_full() -> Result<(), ViewError> {
1139        const N: usize = 4;
1140        let context = MemoryContext::new_for_testing(());
1141        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1142        // Push enough to create several middles, then save.
1143        for i in 0..u32::try_from(5 * N + 2).unwrap() {
1144            view.push_back(i);
1145        }
1146        save(&context, &mut view).await?;
1147
1148        // Push a partial back to verify the next save merges and re-chunks correctly.
1149        view.push_back(1000);
1150        view.push_back(1001);
1151        save(&context, &mut view).await?;
1152
1153        // Drop a few front elements (less than N), save, and check again.
1154        for _ in 0..(N - 1) {
1155            view.delete_front().await?;
1156        }
1157        save(&context, &mut view).await?;
1158
1159        // After all this, every middle bucket in storage must hold exactly N elements.
1160        let view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1161        let first_index = view.stored_first_index;
1162        for offset in 1..view.stored_num_buckets.saturating_sub(1) {
1163            let key = view.get_middle_key(first_index + offset)?;
1164            let data = context
1165                .store()
1166                .read_value::<Vec<u32>>(&key)
1167                .await?
1168                .expect("middle bucket should be present in storage");
1169            assert_eq!(
1170                data.len(),
1171                N,
1172                "middle at offset {offset} should hold N elements"
1173            );
1174        }
1175        Ok(())
1176    }
1177
1178    /// `stored_count` must be exact across the partial-front and partial-back
1179    /// edge cases, even when the cursor has advanced inside the front bucket.
1180    #[tokio::test]
1181    async fn stored_count_is_exact() -> Result<(), ViewError> {
1182        const N: usize = 3;
1183        let context = MemoryContext::new_for_testing(());
1184        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1185        // 7 elements -> front [0,1,2], middle [3,4,5], back [6].
1186        for i in 0..7u32 {
1187            view.push_back(i);
1188        }
1189        save(&context, &mut view).await?;
1190
1191        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1192        assert_eq!(view.stored_count(), 7);
1193        view.delete_front().await?; // drop 0
1194        assert_eq!(view.stored_count(), 6);
1195        view.delete_front().await?; // drop 1
1196        assert_eq!(view.stored_count(), 5);
1197        view.delete_front().await?; // drop 2, crosses into middle bucket
1198        assert_eq!(view.stored_count(), 4);
1199        Ok(())
1200    }
1201
1202    /// Build a view that hits each `SaveCase` variant and verify dispatch +
1203    /// roundtrip. Pinning each case to a concrete scenario means a refactor
1204    /// that silently drops a branch fails loudly instead of relying on the
1205    /// random fuzz to catch it eventually.
1206    #[tokio::test]
1207    async fn save_plan_covers_each_case() -> Result<(), ViewError> {
1208        const N: usize = 3;
1209
1210        // Empty: a freshly-loaded view with no pending changes.
1211        let context = MemoryContext::new_for_testing(());
1212        let view = BucketQueueView::<_, u32, N>::load(context).await?;
1213        assert!(matches!(view.save_plan()?.case, SaveCase::Empty));
1214
1215        // MetadataOnly: a single stored bucket with the cursor advanced inside it.
1216        let context = MemoryContext::new_for_testing(());
1217        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1218        view.push_back(10);
1219        view.push_back(20);
1220        save(&context, &mut view).await?;
1221        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1222        view.delete_front().await?;
1223        assert!(matches!(view.save_plan()?.case, SaveCase::MetadataOnly));
1224
1225        // Rewrite: <= 1 bucket survives but there is a structural change
1226        // (consumed past the front bucket; only the back remains).
1227        let context = MemoryContext::new_for_testing(());
1228        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1229        for i in 0..5u32 {
1230            view.push_back(i);
1231        }
1232        save(&context, &mut view).await?;
1233        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1234        for _ in 0..N {
1235            view.delete_front().await?;
1236        }
1237        assert!(matches!(view.save_plan()?.case, SaveCase::Rewrite));
1238
1239        // Patch: >= 2 buckets survive (here: front + middle + back, with a
1240        // pending new value to force the re-chunk path).
1241        let context = MemoryContext::new_for_testing(());
1242        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1243        for i in 0..7u32 {
1244            view.push_back(i);
1245        }
1246        save(&context, &mut view).await?;
1247        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1248        view.push_back(100);
1249        assert!(matches!(view.save_plan()?.case, SaveCase::Patch { .. }));
1250
1251        Ok(())
1252    }
1253
1254    /// The most intricate single save: in one `Patch` the cursor has crossed a bucket
1255    /// boundary (so the front is promoted from a former middle), one or more middles
1256    /// survive unchanged, *and* enough new values are pending that merging them with the
1257    /// old back re-chunks into several new middles plus a new back. This exercises front
1258    /// promotion, middle key preservation, and the re-chunk path simultaneously — the one
1259    /// combination `save_plan_covers_each_case` only hits piecemeal.
1260    #[tokio::test]
1261    async fn patch_promotes_front_keeps_middles_and_rechunks() -> Result<(), ViewError> {
1262        const N: usize = 3;
1263        let context = MemoryContext::new_for_testing(());
1264        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1265        // 14 elements -> front [0,1,2], middles [3,4,5] [6,7,8] [9,10,11], back [12,13].
1266        for i in 0..14u32 {
1267            view.push_back(i);
1268        }
1269        save(&context, &mut view).await?;
1270
1271        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1272        // Drop 0,1,2,3: the cursor crosses out of the front bucket and lands at value 4
1273        // inside the former middle [3,4,5], which must now be promoted to the front.
1274        for _ in 0..4 {
1275            view.delete_front().await?;
1276        }
1277        // Push 6 values so that merging with the partial back [12,13] yields
1278        // [12,13,100][101,102,103][104,105]: two new middles and a new back.
1279        for v in 100..106u32 {
1280            view.push_back(v);
1281        }
1282
1283        // Confirm we are about to take the Patch path with the front having moved by a
1284        // whole bucket (remaining_offset == 1) and four buckets surviving the cursor.
1285        let plan = view.save_plan()?;
1286        assert_eq!(plan.remaining_offset, 1);
1287        assert!(matches!(
1288            plan.case,
1289            SaveCase::Patch {
1290                remaining_count: 4,
1291                ..
1292            }
1293        ));
1294
1295        save(&context, &mut view).await?;
1296
1297        // Reload from storage and check the full sequence survived the re-chunk.
1298        let view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1299        let expected: Vec<u32> = (4..14).chain(100..106).collect();
1300        assert_eq!(view.elements().await?, expected);
1301        assert_eq!(view.count(), expected.len());
1302
1303        // The layout grew to 6 buckets (front + 4 middles + back) and every middle still
1304        // holds exactly N — the invariant the whole design rests on.
1305        assert_eq!(view.stored_num_buckets, 6);
1306        let first_index = view.stored_first_index;
1307        for offset in 1..view.stored_num_buckets - 1 {
1308            let key = view.get_middle_key(first_index + offset)?;
1309            let data = context
1310                .store()
1311                .read_value::<Vec<u32>>(&key)
1312                .await?
1313                .expect("middle bucket should be present in storage");
1314            assert_eq!(data.len(), N, "middle at offset {offset} must hold N");
1315        }
1316        Ok(())
1317    }
1318
1319    /// N=1 is degenerate: every bucket holds exactly one element, the front
1320    /// and back can't share a bucket, and every middle is also a single
1321    /// element. Exercise enough operations to cross several bucket boundaries.
1322    #[tokio::test]
1323    async fn n_equals_one_roundtrip() -> Result<(), ViewError> {
1324        let context = MemoryContext::new_for_testing(());
1325        let mut view = BucketQueueView::<_, u32, 1>::load(context.clone()).await?;
1326        for i in 0..5u32 {
1327            view.push_back(i);
1328        }
1329        save(&context, &mut view).await?;
1330
1331        let mut view = BucketQueueView::<_, u32, 1>::load(context.clone()).await?;
1332        assert_eq!(view.elements().await?, vec![0, 1, 2, 3, 4]);
1333        view.delete_front().await?;
1334        view.delete_front().await?;
1335        view.push_back(99);
1336        save(&context, &mut view).await?;
1337
1338        let view = BucketQueueView::<_, u32, 1>::load(context).await?;
1339        assert_eq!(view.elements().await?, vec![2, 3, 4, 99]);
1340        Ok(())
1341    }
1342
1343    /// `rollback` must wipe in-memory edits and restore the cursor / state
1344    /// to whatever was last saved.
1345    #[tokio::test]
1346    async fn rollback_restores_saved_state() -> Result<(), ViewError> {
1347        const N: usize = 3;
1348        let context = MemoryContext::new_for_testing(());
1349        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1350        for i in 0..5u32 {
1351            view.push_back(i);
1352        }
1353        save(&context, &mut view).await?;
1354
1355        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1356        view.delete_front().await?;
1357        view.delete_front().await?;
1358        view.push_back(100);
1359        view.push_back(101);
1360        assert_eq!(view.elements().await?, vec![2, 3, 4, 100, 101]);
1361
1362        view.rollback();
1363        assert_eq!(view.elements().await?, vec![0, 1, 2, 3, 4]);
1364
1365        // After rollback, has_pending_changes must be false again.
1366        assert!(!view.has_pending_changes().await);
1367        Ok(())
1368    }
1369
1370    /// `clear()` followed by `rollback()` must restore the last-saved state,
1371    /// including a non-zero saved front position. `rollback` is synchronous and
1372    /// rebuilds the cursor purely from the in-memory saved layout, so `clear` must
1373    /// preserve it.
1374    #[tokio::test]
1375    async fn rollback_after_clear_restores_front_position() -> Result<(), ViewError> {
1376        const N: usize = 5;
1377        let context = MemoryContext::new_for_testing(());
1378        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1379        for value in [10u32, 20, 30] {
1380            view.push_back(value);
1381        }
1382        save(&context, &mut view).await?;
1383
1384        // Advance the saved front past position 0 (single bucket -> MetadataOnly).
1385        let mut view = BucketQueueView::<_, u32, N>::load(context.clone()).await?;
1386        view.delete_front().await?; // drop 10
1387        save(&context, &mut view).await?;
1388
1389        let mut view = BucketQueueView::<_, u32, N>::load(context).await?;
1390        assert_eq!(view.elements().await?, vec![20, 30]);
1391
1392        view.clear();
1393        assert_eq!(view.elements().await?, Vec::<u32>::new());
1394
1395        view.rollback();
1396        assert_eq!(view.elements().await?, vec![20, 30]);
1397        assert!(!view.has_pending_changes().await);
1398        Ok(())
1399    }
1400
1401    async fn save<V: View>(context: &V::Context, view: &mut V) -> Result<(), ViewError> {
1402        let mut batch = Batch::new();
1403        view.pre_save(&mut batch)?;
1404        context.store().write_batch(batch).await?;
1405        view.post_save();
1406        Ok(())
1407    }
1408}
linera_views/views/bucket_queue_view.rs

linera_views/views/
bucket_queue_view.rs
