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

fedimint_eventlog/
lib.rs
