p2panda_stream/log_prune/
processor.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::borrow::Borrow;
4use std::cell::RefCell;
5use std::collections::VecDeque;
6use std::marker::PhantomData;
7
8use p2panda_core::{AnyOperation, Hash, LogId, SeqNum, VerifyingKey};
9use p2panda_store::Transaction;
10use p2panda_store::logs::LogStore;
11use thiserror::Error;
12use tokio::sync::Notify;
13
14use crate::Processor;
15use crate::log_prune::LogPruneArgs;
16
17pub struct LogPrune<S, T, L> {
18    store: S,
19    notify: Notify,
20    queue: RefCell<VecDeque<(T, LogPruneResult)>>,
21    _marker: PhantomData<L>,
22}
23
24impl<S, T, L> LogPrune<S, T, L>
25where
26    S: LogStore<AnyOperation, VerifyingKey, L, SeqNum, Hash>,
27    L: LogId,
28{
29    pub fn new(store: S) -> Self {
30        Self {
31            store,
32            notify: Notify::new(),
33            queue: RefCell::new(VecDeque::new()),
34            _marker: PhantomData,
35        }
36    }
37}
38
39impl<S, T, L> Processor<T> for LogPrune<S, T, L>
40where
41    S: Transaction + LogStore<AnyOperation, VerifyingKey, L, SeqNum, Hash>,
42    T: Borrow<LogPruneArgs<VerifyingKey, L, SeqNum>>,
43    L: LogId,
44{
45    type Output = (T, LogPruneResult);
46
47    type Error = (T, LogPruneError);
48
49    async fn process(&self, input: T) -> Result<(), Self::Error> {
50        let args: &LogPruneArgs<VerifyingKey, L, SeqNum> = input.borrow();
51
52        let result = if let LogPruneArgs::PruneEntriesUntil {
53            author,
54            log_id,
55            seq_num,
56        } = args
57        {
58            let permit = match self.store.begin().await {
59                Ok(permit) => permit,
60                Err(err) => return Err((input, LogPruneError::StoreError(err.to_string()))),
61            };
62
63            match self.store.prune_entries_tx(author, log_id, seq_num).await {
64                Ok(num_entries) => {
65                    if let Err(err) = self.store.commit(permit).await {
66                        return Err((input, LogPruneError::StoreError(err.to_string())));
67                    };
68
69                    (input, LogPruneResult::Pruned { num_entries })
70                }
71                Err(err) => {
72                    // Return the input arguments next to the error to allow mapping it back to
73                    // it's source.
74                    return Err((input, LogPruneError::StoreError(err.to_string())));
75                }
76            }
77        } else {
78            (input, LogPruneResult::Noop)
79        };
80
81        self.queue.borrow_mut().push_back(result);
82        self.notify.notify_one(); // Wake up any pending recv.
83
84        Ok(())
85    }
86
87    async fn next(&self) -> Result<Self::Output, Self::Error> {
88        loop {
89            if let Some(item) = self.queue.borrow_mut().pop_front() {
90                return Ok(item);
91            }
92
93            // Wait for notification that an item was added.
94            self.notify.notified().await;
95        }
96    }
97}
98
99#[derive(Clone, Debug, PartialEq, Eq)]
100pub enum LogPruneResult {
101    Noop,
102    Pruned { num_entries: u64 },
103}
104
105#[derive(Clone, Debug, Error)]
106pub enum LogPruneError {
107    /// Critical storage failure occurred. This is usually a reason to panic.
108    #[error("critical storage failure: {0}")]
109    StoreError(String),
110}
p2panda_stream/log_prune/processor.rs

p2panda_stream/log_prune/
processor.rs
