Repository: meilisearch/meilisearch
File: meilisearch-lib/src/tasks/task_store/store.rs (Rust)

Code before:
}

impl<'a> BytesDecode<'a> for IndexUidTaskIdCodec {
    type DItem = (&'a str, TaskId);

    fn bytes_decode(bytes: &'a [u8]) -> Option<Self::DItem> {
        let len = bytes.len();
        let str_bytes = &bytes[..(len - size_of::<TaskId>() - 1)];
        let str = std::str::from_utf8(str_bytes).ok()?;
        let id = TaskId::from_be_bytes(bytes[(len - size_of::<TaskId>())..].try_into().ok()?);
        Some((str, id))
    }
}

pub struct Store {
    env: Env,
    uids_task_ids: Database<IndexUidTaskIdCodec, Unit>,
    tasks: Database<OwnedType<BEU64>, SerdeJson<Task>>,
}

impl Store {
    /// Create a new store from the specified `Path`.
    /// Be really cautious when calling this function, the returned `Store` is
    /// not altered and is a direct representation of what was laying on the disk.
    /// You probably want to put back all un-finished update in your queue with the
    /// `reset_and_return_unfinished_update` method.
    pub fn new(path: impl AsRef<Path>, size: usize) -> Result<Self> {
        let mut options = EnvOpenOptions::new();
        options.map_size(size);
        options.max_dbs(1000);
        let env = options.open(path)?;

        let uids_task_ids = env.create_database(Some(UID_TASK_IDS))?;
        let tasks = env.create_database(Some(TASKS))?;

        Ok(Self {
            env,
            uids_task_ids,
            tasks,
        })
    }

    /// This function should be called *right after* creating the store.
    /// It put back all unfinished update in the `Created` state. This
    /// allow us to re-enqueue an update that didn't had the time to finish
    /// when Meilisearch closed.
    pub fn reset_and_return_unfinished_tasks(&mut self) -> Result<BinaryHeap<Reverse<TaskId>>> {
        let mut unfinished_tasks: BinaryHeap<Reverse<TaskId>> = BinaryHeap::new();

        let mut wtxn = self.wtxn()?;
        let mut iter = self.tasks.rev_iter_mut(&mut wtxn)?;

Review comment (suggestion):
```suggestion
    /// Be really cautious when calling this function, the returned `Store` may be in an invalid state, with dangling processing tasks.
    /// You will want to patch  all un-finished tasksm and put them in your pending queue with the
    /// `reset_and_return_unfinished_update` method.
```

Code after:
}

impl<'a> BytesDecode<'a> for IndexUidTaskIdCodec {
    type DItem = (&'a str, TaskId);

    fn bytes_decode(bytes: &'a [u8]) -> Option<Self::DItem> {
        let len = bytes.len();
        let str_bytes = &bytes[..(len - size_of::<TaskId>() - 1)];
        let str = std::str::from_utf8(str_bytes).ok()?;
        let id = TaskId::from_be_bytes(bytes[(len - size_of::<TaskId>())..].try_into().ok()?);
        Some((str, id))
    }
}

pub struct Store {
    env: Env,
    uids_task_ids: Database<IndexUidTaskIdCodec, Unit>,
    tasks: Database<OwnedType<BEU64>, SerdeJson<Task>>,
}

impl Store {
    /// Create a new store from the specified `Path`.
    /// Be really cautious when calling this function, the returned `Store` may
    /// be in an invalid state, with dangling processing tasks.
    /// You want to patch  all un-finished tasks and put them in your pending
    /// queue with the `reset_and_return_unfinished_update` method.
    pub fn new(path: impl AsRef<Path>, size: usize) -> Result<Self> {
        let mut options = EnvOpenOptions::new();
        options.map_size(size);
        options.max_dbs(1000);
        let env = options.open(path)?;

        let uids_task_ids = env.create_database(Some(UID_TASK_IDS))?;
        let tasks = env.create_database(Some(TASKS))?;

        Ok(Self {
            env,
            uids_task_ids,
            tasks,
        })
    }

    /// This function should be called *right after* creating the store.
    /// It put back all unfinished update in the `Created` state. This
    /// allow us to re-enqueue an update that didn't had the time to finish
    /// when Meilisearch closed.
    pub fn reset_and_return_unfinished_tasks(&mut self) -> Result<BinaryHeap<Reverse<TaskId>>> {
        let mut unfinished_tasks: BinaryHeap<Reverse<TaskId>> = BinaryHeap::new();

        let mut wtxn = self.wtxn()?;
        let mut iter = self.tasks.rev_iter_mut(&mut wtxn)?;