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::{Extensions, Hash, LogId, Operation, SeqNum, VerifyingKey};
9use p2panda_store::logs::LogStore;
10use thiserror::Error;
11use tokio::sync::Notify;
12
13use crate::Processor;
14use crate::log_prune::LogPruneArgs;
15
16pub struct LogPrune<S, T, L, E> {
17    store: S,
18    notify: Notify,
19    queue: RefCell<VecDeque<(T, LogPruneResult)>>,
20    _marker: PhantomData<(L, E)>,
21}
22
23impl<S, T, L, E> LogPrune<S, T, L, E>
24where
25    S: LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>,
26    L: LogId,
27    E: Extensions,
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, E> Processor<T> for LogPrune<S, T, L, E>
40where
41    S: LogStore<Operation<E>, VerifyingKey, L, SeqNum, Hash>,
42    T: Borrow<LogPruneArgs<VerifyingKey, L, SeqNum>>,
43    L: LogId,
44    E: Extensions,
45{
46    type Output = (T, LogPruneResult);
47
48    type Error = (T, LogPruneError);
49
50    async fn process(&self, input: T) -> Result<(), Self::Error> {
51        let args: &LogPruneArgs<VerifyingKey, L, SeqNum> = input.borrow();
52
53        let result = if let LogPruneArgs::PruneEntriesUntil {
54            author,
55            log_id,
56            seq_num,
57        } = args
58        {
59            match self.store.prune_entries(author, log_id, seq_num).await {
60                Ok(num_entries) => (input, LogPruneResult::Pruned { num_entries }),
61                Err(err) => {
62                    // Return the input arguments next to the error to allow mapping it back to
63                    // it's source.
64                    return Err((input, LogPruneError::StoreError(err.to_string())));
65                }
66            }
67        } else {
68            (input, LogPruneResult::Noop)
69        };
70
71        self.queue.borrow_mut().push_back(result);
72        self.notify.notify_one(); // Wake up any pending recv.
73
74        Ok(())
75    }
76
77    async fn next(&self) -> Result<Self::Output, Self::Error> {
78        loop {
79            if let Some(item) = self.queue.borrow_mut().pop_front() {
80                return Ok(item);
81            }
82
83            // Wait for notification that an item was added.
84            self.notify.notified().await;
85        }
86    }
87}
88
89#[derive(Clone, Debug, PartialEq, Eq)]
90pub enum LogPruneResult {
91    Noop,
92    Pruned { num_entries: u64 },
93}
94
95#[derive(Clone, Debug, Error)]
96pub enum LogPruneError {
97    /// Critical storage failure occurred. This is usually a reason to panic.
98    #[error("critical storage failure: {0}")]
99    StoreError(String),
100}
p2panda_stream/log_prune/processor.rs

p2panda_stream/log_prune/
processor.rs
