fedimint_eventlog/
lib.rs

1#![allow(clippy::needless_lifetimes)]
2
3//! Client Event Log
4//!
5//! The goal here is to maintain a single, ordered, append only
6//! log of all important client-side events: low or high level,
7//! and move as much of coordination between different parts of
8//! the system in a natural and decomposed way.
9//!
10//! Any event log "follower" can just keep going through
11//! all events and react to ones it is interested in (and understands),
12//! potentially emitting events of its own, and atomically updating persisted
13//! event log position ("cursor") of events that were already processed.
14
15#[cfg(feature = "uniffi")]
16::uniffi::setup_scaffolding!();
17
18use std::borrow::Cow;
19use std::str::FromStr;
20use std::sync::atomic::{AtomicU64, Ordering};
21use std::time::Duration;
22use std::{fmt, ops};
23
24use fedimint_core::core::{ModuleInstanceId, ModuleKind};
25use fedimint_core::db::{
26    Database, DatabaseTransaction, IDatabaseTransactionOpsCoreTyped, NonCommittable,
27};
28use fedimint_core::encoding::{Decodable, Encodable};
29use fedimint_core::{Amount, apply, async_trait_maybe_send, impl_db_lookup, impl_db_record};
30use fedimint_logging::LOG_CLIENT_EVENT_LOG;
31use futures::{Future, StreamExt};
32use itertools::Itertools;
33use serde::{Deserialize, Serialize};
34use thiserror::Error;
35use tokio::sync::{broadcast, watch};
36use tracing::{debug, trace};
37
38/// DB prefixes hardcoded for use of the event log
39/// `fedimint-eventlog` was extracted from `fedimint-client` to help
40/// include/re-use in other part of the code. But fundamentally its role
41/// is to implement event log in the client.
42/// There is currently no way to inject the prefixes to use for db records,
43/// so we use these constants to keep them in sync. Any other app that will
44/// want to store its own even log, will need to use the exact same prefixes,
45/// which in practice should not be a problem.
46pub const DB_KEY_PREFIX_UNORDERED_EVENT_LOG: u8 = 0x3a;
47pub const DB_KEY_PREFIX_EVENT_LOG: u8 = 0x39;
48pub const DB_KEY_PREFIX_EVENT_LOG_TRIMABLE: u8 = 0x41;
49
50/// Minimum age in ID count for trimable events to be deleted
51const TRIMABLE_EVENTLOG_MIN_ID_AGE: u64 = 10_000;
52/// Minimum age in microseconds for trimable events to be deleted (14 days)
53const TRIMABLE_EVENTLOG_MIN_TS_AGE: u64 = 14 * 24 * 60 * 60 * 1_000_000;
54/// Maximum number of entries to trim in one operation
55const TRIMABLE_EVENTLOG_MAX_TRIMMED_EVENTS: usize = 100_000;
56
57/// Type of persistence the [`Event`] uses.
58///
59/// As a compromise between richness of events and amount of data to store
60/// Fedimint maintains two event logs in parallel:
61///
62/// * untrimable
63/// * trimable
64///
65/// Untrimable log will append only a subset of events that are infrequent,
66/// but important enough to be forever useful, e.g. for processing or debugging
67/// of historical events.
68///
69/// Trimable log will append all persistent events, but will over time remove
70/// the oldest ones. It will always retain enough events, that no log follower
71/// actively processing it should ever miss any event, but restarting processing
72/// from the start (index 0) can't be used for processing historical data.
73///
74/// Notably the positions in both logs are not interchangeable, so they use
75/// different types.
76///
77/// On top of it, some events are transient and are not persisted at all,
78/// and emitted only at runtime.
79///
80/// Consult [`Event::PERSISTENCE`] to know which event uses which persistence.
81pub enum EventPersistence {
82    /// Not written anywhere, just broadcasted as notification at runtime
83    Transient,
84    /// Persised only to log that gets trimmed
85    Trimable,
86    /// Persisted in both trimmed and untrimmed logs, so potentially
87    /// stored forever.
88    Persistent,
89}
90
91pub trait Event: serde::Serialize + serde::de::DeserializeOwned {
92    const MODULE: Option<ModuleKind>;
93    const KIND: EventKind;
94    const PERSISTENCE: EventPersistence;
95}
96
97/// An counter that resets on every restart, that guarantees that
98/// [`UnordedEventLogId`]s don't conflict with each other.
99static UNORDEREDED_EVENT_LOG_ID_COUNTER: AtomicU64 = AtomicU64::new(0);
100
101/// A self-allocated ID that is mostly ordered
102///
103/// The goal here is to avoid concurrent database transaction
104/// conflicts due the ID allocation. Instead they are picked based on
105/// a time and a counter, so they are mostly but not strictly ordered and
106/// monotonic, and even more importantly: not contiguous.
107#[derive(Debug, Encodable, Decodable)]
108pub struct UnordedEventLogId {
109    ts_usecs: u64,
110    counter: u64,
111}
112
113impl UnordedEventLogId {
114    fn new() -> Self {
115        Self {
116            ts_usecs: u64::try_from(fedimint_core::time::duration_since_epoch().as_micros())
117                // This will never happen
118                .unwrap_or(u64::MAX),
119            counter: UNORDEREDED_EVENT_LOG_ID_COUNTER.fetch_add(1, Ordering::Relaxed),
120        }
121    }
122}
123
124/// Ordered, contiguous ID space, which is easy for event log followers to
125/// track.
126#[derive(
127    Copy,
128    Clone,
129    Debug,
130    Encodable,
131    Decodable,
132    Default,
133    PartialEq,
134    Eq,
135    PartialOrd,
136    Ord,
137    Serialize,
138    Deserialize,
139)]
140pub struct EventLogId(u64);
141
142#[cfg(feature = "uniffi")]
143uniffi::custom_newtype!(EventLogId, u64);
144
145impl EventLogId {
146    pub const LOG_START: EventLogId = EventLogId(0);
147
148    pub fn next(self) -> EventLogId {
149        Self(self.0 + 1)
150    }
151
152    pub fn saturating_add(self, rhs: u64) -> EventLogId {
153        Self(self.0.saturating_add(rhs))
154    }
155
156    pub fn saturating_sub(self, rhs: u64) -> EventLogId {
157        Self(self.0.saturating_sub(rhs))
158    }
159
160    pub fn checked_sub(self, rhs: u64) -> Option<EventLogId> {
161        self.0.checked_sub(rhs).map(EventLogId)
162    }
163}
164
165impl From<EventLogId> for u64 {
166    fn from(value: EventLogId) -> Self {
167        value.0
168    }
169}
170
171impl FromStr for EventLogId {
172    type Err = <u64 as FromStr>::Err;
173
174    fn from_str(s: &str) -> Result<Self, Self::Err> {
175        u64::from_str(s).map(Self)
176    }
177}
178
179impl fmt::Display for EventLogId {
180    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
181        write!(f, "{}", self.0)
182    }
183}
184
185#[derive(Debug, Clone, Encodable, Decodable, PartialEq, Eq, Serialize, Deserialize)]
186pub struct EventKind(Cow<'static, str>);
187
188#[cfg(feature = "uniffi")]
189uniffi::custom_type!(EventKind, String);
190
191impl EventKind {
192    pub const fn from_static(value: &'static str) -> Self {
193        Self(Cow::Borrowed(value))
194    }
195}
196
197impl<'s> From<&'s str> for EventKind {
198    fn from(value: &'s str) -> Self {
199        Self(Cow::Owned(value.to_owned()))
200    }
201}
202
203impl From<String> for EventKind {
204    fn from(value: String) -> Self {
205        Self(Cow::Owned(value))
206    }
207}
208
209impl From<EventKind> for String {
210    fn from(event_kind: EventKind) -> Self {
211        event_kind.0.into_owned()
212    }
213}
214
215impl fmt::Display for EventKind {
216    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
217        f.write_str(&self.0)
218    }
219}
220
221#[derive(Debug, Encodable, Decodable, Clone, Serialize, Deserialize)]
222#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
223pub struct EventLogModule {
224    pub kind: ModuleKind,
225    pub id: ModuleInstanceId,
226}
227
228#[derive(Debug, Encodable, Decodable, Clone)]
229pub struct UnorderedEventLogEntry {
230    pub flags: u8,
231    pub inner: EventLogEntry,
232}
233
234impl UnorderedEventLogEntry {
235    pub const FLAG_PERSIST: u8 = 1;
236    pub const FLAG_TRIMABLE: u8 = 2;
237
238    fn persist(&self) -> bool {
239        self.flags & Self::FLAG_PERSIST != 0
240    }
241
242    fn trimable(&self) -> bool {
243        self.flags & Self::FLAG_TRIMABLE != 0
244    }
245}
246
247#[derive(Debug, Encodable, Decodable, Clone)]
248#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
249pub struct EventLogEntry {
250    /// Type/kind of the event
251    ///
252    /// Any part of the client is free to self-allocate identifier, denoting a
253    /// certain kind of an event. Notably one event kind have multiple
254    /// instances. E.g. "successful wallet deposit" can be an event kind,
255    /// and it can happen multiple times with different payloads.
256    pub kind: EventKind,
257
258    /// To prevent accidental conflicts between `kind`s, a module kind the
259    /// given event kind belong is used as well.
260    ///
261    /// Note: the meaning of this field is mostly about which part of the code
262    /// defines this event kind. Oftentime a core (non-module)-defined event
263    /// will refer in some way to a module. It should use a separate `module_id`
264    /// field in the `payload`, instead of this field.
265    pub module: Option<EventLogModule>,
266
267    /// Timestamp in microseconds after unix epoch
268    pub ts_usecs: u64,
269
270    /// Event-kind specific payload, typically encoded as a json string for
271    /// flexibility.
272    pub payload: Vec<u8>,
273}
274
275impl EventLogEntry {
276    pub fn module_kind(&self) -> Option<&ModuleKind> {
277        self.module.as_ref().map(|m| &m.kind)
278    }
279
280    pub fn module_id(&self) -> Option<ModuleInstanceId> {
281        self.module.as_ref().map(|m| m.id)
282    }
283
284    /// Get the event payload as typed value
285    pub fn to_event<E>(&self) -> Option<E>
286    where
287        E: Event,
288    {
289        serde_json::from_slice(&self.payload).ok()
290    }
291}
292
293/// An `EventLogEntry` that was already persisted (so has an id)
294#[derive(Debug, Clone)]
295#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
296pub struct PersistedLogEntry {
297    id: EventLogId,
298    inner: EventLogEntry,
299}
300
301impl Serialize for PersistedLogEntry {
302    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
303    where
304        S: serde::Serializer,
305    {
306        use serde::ser::SerializeStruct;
307
308        let mut state = serializer.serialize_struct("PersistedLogEntry", 5)?;
309        state.serialize_field("id", &self.id)?;
310        state.serialize_field("kind", &self.inner.kind)?;
311        state.serialize_field("module", &self.inner.module)?;
312        state.serialize_field("ts_usecs", &self.inner.ts_usecs)?;
313
314        // Try to deserialize payload as JSON, fall back to hex encoding
315        let payload_value: serde_json::Value = serde_json::from_slice(&self.inner.payload)
316            .unwrap_or_else(|_| serde_json::Value::String(hex::encode(&self.inner.payload)));
317        state.serialize_field("payload", &payload_value)?;
318
319        state.end()
320    }
321}
322
323impl<'de> Deserialize<'de> for PersistedLogEntry {
324    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
325    where
326        D: serde::Deserializer<'de>,
327    {
328        use serde::de::{self, MapAccess, Visitor};
329
330        #[derive(Deserialize)]
331        #[serde(field_identifier, rename_all = "snake_case")]
332        enum Field {
333            Id,
334            Kind,
335            Module,
336            TsUsecs,
337            Payload,
338        }
339
340        struct PersistedLogEntryVisitor;
341
342        impl<'de> Visitor<'de> for PersistedLogEntryVisitor {
343            type Value = PersistedLogEntry;
344
345            fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
346                formatter.write_str("struct PersistedLogEntry")
347            }
348
349            fn visit_map<V>(self, mut map: V) -> Result<PersistedLogEntry, V::Error>
350            where
351                V: MapAccess<'de>,
352            {
353                let mut id = None;
354                let mut kind = None;
355                let mut module = None;
356                let mut ts_usecs = None;
357                let mut payload = None;
358
359                while let Some(key) = map.next_key()? {
360                    match key {
361                        Field::Id => {
362                            if id.is_some() {
363                                return Err(de::Error::duplicate_field("id"));
364                            }
365                            id = Some(map.next_value()?);
366                        }
367                        Field::Kind => {
368                            if kind.is_some() {
369                                return Err(de::Error::duplicate_field("kind"));
370                            }
371                            kind = Some(map.next_value()?);
372                        }
373                        Field::Module => {
374                            if module.is_some() {
375                                return Err(de::Error::duplicate_field("module"));
376                            }
377                            module = Some(map.next_value()?);
378                        }
379                        Field::TsUsecs => {
380                            if ts_usecs.is_some() {
381                                return Err(de::Error::duplicate_field("ts_usecs"));
382                            }
383                            ts_usecs = Some(map.next_value()?);
384                        }
385                        Field::Payload => {
386                            if payload.is_some() {
387                                return Err(de::Error::duplicate_field("payload"));
388                            }
389                            let value: serde_json::Value = map.next_value()?;
390                            payload = Some(serde_json::to_vec(&value).map_err(de::Error::custom)?);
391                        }
392                    }
393                }
394
395                let id = id.ok_or_else(|| de::Error::missing_field("id"))?;
396                let kind = kind.ok_or_else(|| de::Error::missing_field("kind"))?;
397                let module = module.ok_or_else(|| de::Error::missing_field("module"))?;
398                let ts_usecs = ts_usecs.ok_or_else(|| de::Error::missing_field("ts_usecs"))?;
399                let payload = payload.ok_or_else(|| de::Error::missing_field("payload"))?;
400
401                Ok(PersistedLogEntry {
402                    id,
403                    inner: EventLogEntry {
404                        kind,
405                        module,
406                        ts_usecs,
407                        payload,
408                    },
409                })
410            }
411        }
412
413        const FIELDS: &[&str] = &["id", "kind", "module", "ts_usecs", "payload"];
414        deserializer.deserialize_struct("PersistedLogEntry", FIELDS, PersistedLogEntryVisitor)
415    }
416}
417
418impl PersistedLogEntry {
419    pub fn id(&self) -> EventLogId {
420        self.id
421    }
422
423    pub fn as_raw(&self) -> &EventLogEntry {
424        &self.inner
425    }
426}
427
428impl ops::Deref for PersistedLogEntry {
429    type Target = EventLogEntry;
430
431    fn deref(&self) -> &Self::Target {
432        &self.inner
433    }
434}
435
436impl_db_record!(
437    key = UnordedEventLogId,
438    value = UnorderedEventLogEntry,
439    db_prefix = DB_KEY_PREFIX_UNORDERED_EVENT_LOG,
440);
441
442#[derive(Clone, Debug, Encodable, Decodable)]
443pub struct UnorderedEventLogIdPrefixAll;
444
445impl_db_lookup!(
446    key = UnordedEventLogId,
447    query_prefix = UnorderedEventLogIdPrefixAll
448);
449
450#[derive(Clone, Debug, Encodable, Decodable)]
451pub struct EventLogIdPrefixAll;
452
453#[derive(Clone, Debug, Encodable, Decodable)]
454pub struct EventLogIdPrefix(EventLogId);
455
456impl_db_record!(
457    key = EventLogId,
458    value = EventLogEntry,
459    db_prefix = DB_KEY_PREFIX_EVENT_LOG,
460);
461
462impl_db_lookup!(key = EventLogId, query_prefix = EventLogIdPrefixAll);
463
464impl_db_lookup!(key = EventLogId, query_prefix = EventLogIdPrefix);
465
466#[derive(
467    Copy,
468    Clone,
469    Debug,
470    Encodable,
471    Decodable,
472    Default,
473    PartialEq,
474    Eq,
475    PartialOrd,
476    Ord,
477    Serialize,
478    Deserialize,
479)]
480pub struct EventLogTrimableId(EventLogId);
481
482impl EventLogTrimableId {
483    fn next(&self) -> Self {
484        Self(self.0.next())
485    }
486
487    pub fn saturating_add(self, rhs: u64) -> Self {
488        Self(self.0.saturating_add(rhs))
489    }
490}
491
492impl From<u64> for EventLogTrimableId {
493    fn from(value: u64) -> Self {
494        Self(EventLogId(value))
495    }
496}
497
498#[derive(Clone, Debug, Encodable, Decodable)]
499pub struct EventLogTrimableIdPrefixAll;
500
501#[derive(Clone, Debug, Encodable, Decodable)]
502pub struct EventLogTrimableIdPrefix(EventLogId);
503
504impl_db_record!(
505    key = EventLogTrimableId,
506    value = EventLogEntry,
507    db_prefix = DB_KEY_PREFIX_EVENT_LOG_TRIMABLE,
508);
509
510impl_db_lookup!(
511    key = EventLogTrimableId,
512    query_prefix = EventLogTrimableIdPrefixAll
513);
514
515impl_db_lookup!(
516    key = EventLogTrimableId,
517    query_prefix = EventLogTrimableIdPrefix
518);
519
520#[apply(async_trait_maybe_send!)]
521pub trait DBTransactionEventLogExt {
522    #[allow(clippy::too_many_arguments)]
523    async fn log_event_raw(
524        &mut self,
525        log_ordering_wakeup_tx: watch::Sender<()>,
526        kind: EventKind,
527        module_kind: Option<ModuleKind>,
528        module_id: Option<ModuleInstanceId>,
529        payload: Vec<u8>,
530        persist: EventPersistence,
531    );
532
533    /// Log an event log event
534    ///
535    /// The event will start "unordered", but after it is committed an ordering
536    /// task will be notified to "order" it into a final ordered log.
537    async fn log_event<E>(
538        &mut self,
539        log_ordering_wakeup_tx: watch::Sender<()>,
540        module_id: Option<ModuleInstanceId>,
541        event: E,
542    ) where
543        E: Event + Send,
544    {
545        self.log_event_raw(
546            log_ordering_wakeup_tx,
547            E::KIND,
548            E::MODULE,
549            module_id,
550            serde_json::to_vec(&event).expect("Serialization can't fail"),
551            <E as Event>::PERSISTENCE,
552        )
553        .await;
554    }
555
556    /// Next [`EventLogId`] to use for new ordered events.
557    ///
558    /// Used by ordering task, though might be
559    /// useful to get the current count of events.
560    async fn get_next_event_log_id(&mut self) -> EventLogId;
561
562    /// Next [`EventLogTrimableId`] to use for new ordered trimable events
563    async fn get_next_event_log_trimable_id(&mut self) -> EventLogTrimableId;
564
565    /// Read a part of the event log.
566    async fn get_event_log(
567        &mut self,
568        pos: Option<EventLogId>,
569        limit: u64,
570    ) -> Vec<PersistedLogEntry>;
571
572    async fn get_event_log_trimable(
573        &mut self,
574        pos: Option<EventLogTrimableId>,
575        limit: u64,
576    ) -> Vec<PersistedLogEntry>;
577}
578
579#[apply(async_trait_maybe_send!)]
580impl<'tx, Cap> DBTransactionEventLogExt for DatabaseTransaction<'tx, Cap>
581where
582    Cap: Send,
583{
584    async fn log_event_raw(
585        &mut self,
586        log_ordering_wakeup_tx: watch::Sender<()>,
587        kind: EventKind,
588        module_kind: Option<ModuleKind>,
589        module_id: Option<ModuleInstanceId>,
590        payload: Vec<u8>,
591        persist: EventPersistence,
592    ) {
593        assert_eq!(
594            module_kind.is_some(),
595            module_id.is_some(),
596            "Events of modules must have module_id set"
597        );
598
599        let unordered_id = UnordedEventLogId::new();
600        trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, "New unordered event log event");
601
602        if self
603            .insert_entry(
604                &unordered_id,
605                &UnorderedEventLogEntry {
606                    flags: match persist {
607                        EventPersistence::Transient => 0,
608                        EventPersistence::Trimable => UnorderedEventLogEntry::FLAG_TRIMABLE,
609                        EventPersistence::Persistent => UnorderedEventLogEntry::FLAG_PERSIST,
610                    },
611                    inner: EventLogEntry {
612                        kind,
613                        module: module_kind.map(|kind| EventLogModule {
614                            kind,
615                            id: module_id.expect("module_id exists for module events"),
616                        }),
617                        ts_usecs: unordered_id.ts_usecs,
618                        payload,
619                    },
620                },
621            )
622            .await
623            .is_some()
624        {
625            panic!("Trying to overwrite event in the client event log");
626        }
627        self.on_commit(move || {
628            log_ordering_wakeup_tx.send_replace(());
629        });
630    }
631
632    async fn get_next_event_log_id(&mut self) -> EventLogId {
633        self.find_by_prefix_sorted_descending(&EventLogIdPrefixAll)
634            .await
635            .next()
636            .await
637            .map(|(k, _v)| k.next())
638            .unwrap_or_default()
639    }
640
641    async fn get_next_event_log_trimable_id(&mut self) -> EventLogTrimableId {
642        EventLogTrimableId(
643            self.find_by_prefix_sorted_descending(&EventLogTrimableIdPrefixAll)
644                .await
645                .next()
646                .await
647                .map(|(k, _v)| k.0.next())
648                .unwrap_or_default(),
649        )
650    }
651
652    async fn get_event_log(
653        &mut self,
654        pos: Option<EventLogId>,
655        limit: u64,
656    ) -> Vec<PersistedLogEntry> {
657        let pos = pos.unwrap_or_default();
658        self.find_by_range(pos..pos.saturating_add(limit))
659            .await
660            .map(|(k, v)| PersistedLogEntry { id: k, inner: v })
661            .collect()
662            .await
663    }
664
665    async fn get_event_log_trimable(
666        &mut self,
667        pos: Option<EventLogTrimableId>,
668        limit: u64,
669    ) -> Vec<PersistedLogEntry> {
670        let pos = pos.unwrap_or_default();
671        self.find_by_range(pos..pos.saturating_add(limit))
672            .await
673            .map(|(k, v)| PersistedLogEntry { id: k.0, inner: v })
674            .collect()
675            .await
676    }
677}
678
679/// Trims old entries from the trimable event log
680async fn trim_trimable_log(db: &Database, current_time_usecs: u64) {
681    let mut dbtx = db.begin_transaction().await;
682
683    let current_trimable_id = dbtx.get_next_event_log_trimable_id().await;
684    let min_id_threshold = current_trimable_id
685        .0
686        .saturating_sub(TRIMABLE_EVENTLOG_MIN_ID_AGE);
687    let min_ts_threshold = current_time_usecs.saturating_sub(TRIMABLE_EVENTLOG_MIN_TS_AGE);
688
689    let entries_to_delete: Vec<_> = dbtx
690        .find_by_prefix(&EventLogTrimableIdPrefixAll)
691        .await
692        .take_while(|(id, entry)| {
693            let id_old_enough = id.0 <= min_id_threshold;
694            let ts_old_enough = entry.ts_usecs <= min_ts_threshold;
695
696            // Continue while both conditions are met
697            async move { id_old_enough && ts_old_enough }
698        })
699        .take(TRIMABLE_EVENTLOG_MAX_TRIMMED_EVENTS)
700        .map(|(id, _entry)| id)
701        .collect()
702        .await;
703
704    for id in &entries_to_delete {
705        dbtx.remove_entry(id).await;
706    }
707
708    dbtx.commit_tx().await;
709}
710
711/// The code that handles new unordered events and rewriters them fully ordered
712/// into the final event log.
713pub async fn run_event_log_ordering_task(
714    db: Database,
715    mut log_ordering_task_wakeup: watch::Receiver<()>,
716    log_event_added: watch::Sender<()>,
717    log_event_added_transient: broadcast::Sender<EventLogEntry>,
718) {
719    debug!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task started");
720
721    let current_time_usecs =
722        u64::try_from(fedimint_core::time::duration_since_epoch().as_micros()).unwrap_or(u64::MAX);
723    trim_trimable_log(&db, current_time_usecs).await;
724
725    let mut next_entry_id = db
726        .begin_transaction_nc()
727        .await
728        .get_next_event_log_id()
729        .await;
730    let mut next_entry_id_trimable = db
731        .begin_transaction_nc()
732        .await
733        .get_next_event_log_trimable_id()
734        .await;
735
736    loop {
737        let mut dbtx = db.begin_transaction().await;
738
739        let unordered_events = dbtx
740            .find_by_prefix(&UnorderedEventLogIdPrefixAll)
741            .await
742            .collect::<Vec<_>>()
743            .await;
744        trace!(target: LOG_CLIENT_EVENT_LOG, num=unordered_events.len(), "Fetched unordered events");
745
746        for (unordered_id, entry) in &unordered_events {
747            assert!(
748                dbtx.remove_entry(unordered_id).await.is_some(),
749                "Must never fail to remove entry"
750            );
751            if entry.persist() {
752                // Non-trimable events get persisted in both the default event log
753                // and trimable event log
754                if !entry.trimable() {
755                    assert!(
756                        dbtx.insert_entry(&next_entry_id, &entry.inner)
757                            .await
758                            .is_none(),
759                        "Must never overwrite existing event"
760                    );
761                    trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Ordered event log event");
762                    next_entry_id = next_entry_id.next();
763                }
764
765                // Trimable events get persisted only in trimable log
766                assert!(
767                    dbtx.insert_entry(&next_entry_id_trimable, &entry.inner)
768                        .await
769                        .is_none(),
770                    "Must never overwrite existing event"
771                );
772                trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Ordered event log event");
773                next_entry_id_trimable = next_entry_id_trimable.next();
774            } else {
775                // Transient events don't get persisted at all
776                trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Transient event log event");
777                dbtx.on_commit({
778                    let log_event_added_transient = log_event_added_transient.clone();
779                    let entry = entry.inner.clone();
780
781                    move || {
782                        // we ignore the no-subscribers
783                        let _ = log_event_added_transient.send(entry);
784                    }
785                });
786            }
787        }
788
789        // This thread is the only thread deleting already existing element of unordered
790        // log and inserting new elements into ordered log, so it should never
791        // fail to commit.
792        dbtx.commit_tx().await;
793        if !unordered_events.is_empty() {
794            log_event_added.send_replace(());
795        }
796
797        trace!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task waits for more events");
798        if log_ordering_task_wakeup.changed().await.is_err() {
799            break;
800        }
801    }
802
803    debug!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task finished");
804}
805
806/// Persistent tracker of a position in the event log
807///
808/// During processing of event log the downstream consumer needs to
809/// keep track of which event were processed already. It needs to do it
810/// atomically and persist it so event in the presence of crashes no
811/// event is ever missed or processed twice.
812///
813/// This trait allows abstracting away where and how is such position stored,
814/// e.g. which key exactly is used, in what prefixed namespace etc.
815///
816/// ## Trimmable vs Non-Trimable log
817///
818/// See [`EventPersistence`]
819#[apply(async_trait_maybe_send!)]
820pub trait EventLogNonTrimableTracker {
821    // Store position in the event log
822    async fn store(
823        &mut self,
824        dbtx: &mut DatabaseTransaction<NonCommittable>,
825        pos: EventLogId,
826    ) -> Result<(), EventLogTrackerError>;
827
828    /// Load the last previous stored position (or None if never stored)
829    async fn load(
830        &mut self,
831        dbtx: &mut DatabaseTransaction<NonCommittable>,
832    ) -> Result<Option<EventLogId>, EventLogTrackerError>;
833}
834pub type DynEventLogTracker = Box<dyn EventLogNonTrimableTracker>;
835
836/// Like [`EventLogNonTrimableTracker`] but for trimable event log
837#[apply(async_trait_maybe_send!)]
838pub trait EventLogTrimableTracker {
839    // Store position in the event log
840    async fn store(
841        &mut self,
842        dbtx: &mut DatabaseTransaction<NonCommittable>,
843        pos: EventLogTrimableId,
844    ) -> Result<(), EventLogTrackerError>;
845
846    /// Load the last previous stored position (or None if never stored)
847    async fn load(
848        &mut self,
849        dbtx: &mut DatabaseTransaction<NonCommittable>,
850    ) -> Result<Option<EventLogTrimableId>, EventLogTrackerError>;
851}
852pub type DynEventLogTrimableTracker = Box<dyn EventLogTrimableTracker>;
853
854/// A failure of an event log tracker to load or store its position in the
855/// log.
856///
857/// Trackers are implemented by applications, so the causes are their own.
858/// This type carries them unchanged: its `Display` and its `source()` are the
859/// cause's.
860#[derive(Debug, Error)]
861#[error(transparent)]
862pub struct EventLogTrackerError(Box<dyn std::error::Error + Send + Sync>);
863
864impl EventLogTrackerError {
865    /// Wraps a failure of an event log tracker, which may be any error value
866    /// or a plain message.
867    pub fn new<E>(source: E) -> Self
868    where
869        E: Into<Box<dyn std::error::Error + Send + Sync>>,
870    {
871        Self(source.into())
872    }
873}
874
875/// A failure while handling entries from the event log.
876#[derive(Debug, Error)]
877#[non_exhaustive]
878pub enum EventHandlerError<E> {
879    /// The handler rejected an event.
880    #[error("The event handler failed")]
881    Handler(#[source] E),
882
883    /// The tracker could not load or store the position in the log.
884    #[error("The event log tracker failed")]
885    Tracker(#[source] Box<dyn std::error::Error + Send + Sync>),
886}
887
888/// Feeds every entry of the event log, past and future, to `call_fn`.
889///
890/// `tracker` remembers how far the log has been read so a restart resumes where
891/// it left off; the loop stops when `call_fn` fails or the client shuts down.
892pub async fn handle_events<F, R, E>(
893    db: Database,
894    mut tracker: DynEventLogTracker,
895    mut log_event_added: watch::Receiver<()>,
896    call_fn: F,
897) -> Result<(), EventHandlerError<E>>
898where
899    F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
900    R: Future<Output = Result<(), E>>,
901    E: std::error::Error + 'static,
902{
903    let mut next_key: EventLogId = tracker
904        .load(&mut db.begin_transaction_nc().await)
905        .await
906        .map_err(|err| EventHandlerError::Tracker(err.into()))?
907        .unwrap_or_default();
908
909    trace!(target: LOG_CLIENT_EVENT_LOG, ?next_key, "Handling events");
910
911    loop {
912        let mut dbtx = db.begin_transaction().await;
913
914        match dbtx.get_value(&next_key).await {
915            Some(event) => {
916                (call_fn)(&mut dbtx.to_ref_nc(), event)
917                    .await
918                    .map_err(EventHandlerError::Handler)?;
919
920                next_key = next_key.next();
921
922                tracker
923                    .store(&mut dbtx.to_ref_nc(), next_key)
924                    .await
925                    .map_err(|err| EventHandlerError::Tracker(err.into()))?;
926
927                dbtx.commit_tx().await;
928            }
929            _ => {
930                if log_event_added.changed().await.is_err() {
931                    break Ok(());
932                }
933            }
934        }
935    }
936}
937
938/// Like [`handle_events`], for the trimable part of the log: entries are handed
939/// to `call_fn` and the tracker's position drives when they can be trimmed.
940pub async fn handle_trimable_events<F, R, E>(
941    db: Database,
942    mut tracker: DynEventLogTrimableTracker,
943    mut log_event_added: watch::Receiver<()>,
944    call_fn: F,
945) -> Result<(), EventHandlerError<E>>
946where
947    F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
948    R: Future<Output = Result<(), E>>,
949    E: std::error::Error + 'static,
950{
951    let mut next_key: EventLogTrimableId = tracker
952        .load(&mut db.begin_transaction_nc().await)
953        .await
954        .map_err(|err| EventHandlerError::Tracker(err.into()))?
955        .unwrap_or_default();
956    trace!(target: LOG_CLIENT_EVENT_LOG, ?next_key, "Handling trimable events");
957
958    loop {
959        let mut dbtx = db.begin_transaction().await;
960
961        match dbtx.get_value(&next_key).await {
962            Some(event) => {
963                (call_fn)(&mut dbtx.to_ref_nc(), event)
964                    .await
965                    .map_err(EventHandlerError::Handler)?;
966
967                next_key = next_key.next();
968                tracker
969                    .store(&mut dbtx.to_ref_nc(), next_key)
970                    .await
971                    .map_err(|err| EventHandlerError::Tracker(err.into()))?;
972
973                dbtx.commit_tx().await;
974            }
975            _ => {
976                if log_event_added.changed().await.is_err() {
977                    break Ok(());
978                }
979            }
980        }
981    }
982}
983
984/// Filters the `PersistedLogEntries` by the `EventKind` and
985/// `ModuleKind`.
986pub fn filter_events_by_kind<'a, I>(
987    all_events: I,
988    module_kind: ModuleKind,
989    event_kind: EventKind,
990) -> impl Iterator<Item = &'a PersistedLogEntry> + 'a
991where
992    I: IntoIterator<Item = &'a PersistedLogEntry> + 'a,
993{
994    all_events.into_iter().filter(move |e| {
995        if let Some(module) = &e.inner.module {
996            e.inner.kind == event_kind && module.kind == module_kind
997        } else {
998            false
999        }
1000    })
1001}
1002
1003/// Joins two sets of events on a predicate.
1004///
1005/// This function computes a "nested loop join" by first computing the cross
1006/// product of the start event vector and the success/failure event vectors. The
1007/// resulting cartesian product is then filtered according to the join predicate
1008/// supplied in the parameters.
1009///
1010/// This function is intended for small data sets. If the data set relations
1011/// grow, this function should implement a different join algorithm or be moved
1012/// out of the gateway.
1013pub fn join_events<'a, L, R, Res>(
1014    events_l: &'a [&PersistedLogEntry],
1015    events_r: &'a [&PersistedLogEntry],
1016    max_time_distance: Option<Duration>,
1017    predicate: impl Fn(L, R, Duration) -> Option<Res> + 'a,
1018) -> impl Iterator<Item = Res> + 'a
1019where
1020    L: Event,
1021    R: Event,
1022{
1023    events_l
1024        .iter()
1025        .cartesian_product(events_r)
1026        .filter_map(move |(l, r)| {
1027            if L::MODULE.as_ref() == l.as_raw().module_kind()
1028                && L::KIND == l.as_raw().kind
1029                && R::MODULE.as_ref() == r.as_raw().module_kind()
1030                && R::KIND == r.as_raw().kind
1031                && let Some(latency_usecs) = r.inner.ts_usecs.checked_sub(l.inner.ts_usecs)
1032                && max_time_distance.is_none_or(|max| u128::from(latency_usecs) <= max.as_millis())
1033                && let Some(l) = l.as_raw().to_event()
1034                && let Some(r) = r.as_raw().to_event()
1035            {
1036                predicate(l, r, Duration::from_millis(latency_usecs))
1037            } else {
1038                None
1039            }
1040        })
1041}
1042
1043/// Helper struct for storing computed data about outgoing and incoming
1044/// payments.
1045#[derive(Debug, Default)]
1046pub struct StructuredPaymentEvents {
1047    pub latencies_usecs: Vec<u64>,
1048    pub fees: Vec<Amount>,
1049    pub latencies_failure: Vec<u64>,
1050}
1051
1052impl StructuredPaymentEvents {
1053    pub fn new(
1054        success_stats: &[(u64, Amount)],
1055        failure_stats: Vec<u64>,
1056    ) -> StructuredPaymentEvents {
1057        let mut events = StructuredPaymentEvents {
1058            latencies_usecs: success_stats.iter().map(|(l, _)| *l).collect(),
1059            fees: success_stats.iter().map(|(_, f)| *f).collect(),
1060            latencies_failure: failure_stats,
1061        };
1062        events.sort();
1063        events
1064    }
1065
1066    /// Combines this `StructuredPaymentEvents` with the `other`
1067    /// `StructuredPaymentEvents` by appending all of the internal vectors.
1068    pub fn combine(&mut self, other: &mut StructuredPaymentEvents) {
1069        self.latencies_usecs.append(&mut other.latencies_usecs);
1070        self.fees.append(&mut other.fees);
1071        self.latencies_failure.append(&mut other.latencies_failure);
1072        self.sort();
1073    }
1074
1075    /// Sorts this `StructuredPaymentEvents` by sorting all of the internal
1076    /// vectors.
1077    fn sort(&mut self) {
1078        self.latencies_usecs.sort_unstable();
1079        self.fees.sort_unstable();
1080        self.latencies_failure.sort_unstable();
1081    }
1082}
1083
1084#[cfg(test)]
1085mod tests;
fedimint_eventlog/lib.rs

fedimint_eventlog/
lib.rs
