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

fedimint_eventlog/
lib.rs
