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 => {
592                            UnorderedEventLogEntry::FLAG_PERSIST
593                                | UnorderedEventLogEntry::FLAG_TRIMABLE
594                        }
595                        EventPersistence::Persistent => UnorderedEventLogEntry::FLAG_PERSIST,
596                    },
597                    inner: EventLogEntry {
598                        kind,
599                        module: module_kind.map(|kind| EventLogModule {
600                            kind,
601                            id: module_id.expect("module_id exists for module events"),
602                        }),
603                        ts_usecs: unordered_id.ts_usecs,
604                        payload,
605                    },
606                },
607            )
608            .await
609            .is_some()
610        {
611            panic!("Trying to overwrite event in the client event log");
612        }
613        self.on_commit(move || {
614            log_ordering_wakeup_tx.send_replace(());
615        });
616    }
617
618    async fn get_next_event_log_id(&mut self) -> EventLogId {
619        self.find_by_prefix_sorted_descending(&EventLogIdPrefixAll)
620            .await
621            .next()
622            .await
623            .map(|(k, _v)| k.next())
624            .unwrap_or_default()
625    }
626
627    async fn get_next_event_log_trimable_id(&mut self) -> EventLogTrimableId {
628        EventLogTrimableId(
629            self.find_by_prefix_sorted_descending(&EventLogTrimableIdPrefixAll)
630                .await
631                .next()
632                .await
633                .map(|(k, _v)| k.0.next())
634                .unwrap_or_default(),
635        )
636    }
637
638    async fn get_event_log(
639        &mut self,
640        pos: Option<EventLogId>,
641        limit: u64,
642    ) -> Vec<PersistedLogEntry> {
643        let pos = pos.unwrap_or_default();
644        self.find_by_range(pos..pos.saturating_add(limit))
645            .await
646            .map(|(k, v)| PersistedLogEntry { id: k, inner: v })
647            .collect()
648            .await
649    }
650
651    async fn get_event_log_trimable(
652        &mut self,
653        pos: Option<EventLogTrimableId>,
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.0, inner: v })
660            .collect()
661            .await
662    }
663}
664
665/// Trims old entries from the trimable event log
666async fn trim_trimable_log(db: &Database, current_time_usecs: u64) {
667    let mut dbtx = db.begin_transaction().await;
668
669    let current_trimable_id = dbtx.get_next_event_log_trimable_id().await;
670    let min_id_threshold = current_trimable_id
671        .0
672        .saturating_sub(TRIMABLE_EVENTLOG_MIN_ID_AGE);
673    let min_ts_threshold = current_time_usecs.saturating_sub(TRIMABLE_EVENTLOG_MIN_TS_AGE);
674
675    let entries_to_delete: Vec<_> = dbtx
676        .find_by_prefix(&EventLogTrimableIdPrefixAll)
677        .await
678        .take_while(|(id, entry)| {
679            let id_old_enough = id.0 <= min_id_threshold;
680            let ts_old_enough = entry.ts_usecs <= min_ts_threshold;
681
682            // Continue while both conditions are met
683            async move { id_old_enough && ts_old_enough }
684        })
685        .take(TRIMABLE_EVENTLOG_MAX_TRIMMED_EVENTS)
686        .map(|(id, _entry)| id)
687        .collect()
688        .await;
689
690    for id in &entries_to_delete {
691        dbtx.remove_entry(id).await;
692    }
693
694    dbtx.commit_tx().await;
695}
696
697/// How far before the newest unordered event already ordered the ordering task
698/// starts its next scan.
699///
700/// Every unordered event is deleted once ordered, and a key-value store keeps
701/// the deletion as a tombstone until compaction, so a scan of the whole
702/// unordered prefix walks every event ordered since then and its cost grows
703/// with the client's event volume. Unordered ids are wall-clock based and
704/// transactions commit in any order, so an event can only become visible after
705/// newer ones were ordered by as long as its transaction stayed open; the
706/// margin covers that, the periodic full scan below covers the rest.
707const UNORDERED_SCAN_MARGIN: Duration = Duration::from_secs(60);
708
709/// How often the ordering task scans the whole unordered prefix, so an event
710/// committed later than [`UNORDERED_SCAN_MARGIN`] after newer ones, or written
711/// across a backwards clock step, is still ordered within this bound.
712const UNORDERED_FULL_SCAN_INTERVAL: Duration = Duration::from_secs(60);
713
714/// The code that handles new unordered events and rewriters them fully ordered
715/// into the final event log.
716pub async fn run_event_log_ordering_task(
717    db: Database,
718    mut log_ordering_task_wakeup: watch::Receiver<()>,
719    log_event_added: watch::Sender<()>,
720    log_event_added_transient: broadcast::Sender<EventLogEntry>,
721) {
722    debug!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task started");
723
724    let current_time_usecs =
725        u64::try_from(fedimint_core::time::duration_since_epoch().as_micros()).unwrap_or(u64::MAX);
726    trim_trimable_log(&db, current_time_usecs).await;
727
728    let mut next_entry_id = db
729        .begin_transaction_nc()
730        .await
731        .get_next_event_log_id()
732        .await;
733    let mut next_entry_id_trimable = db
734        .begin_transaction_nc()
735        .await
736        .get_next_event_log_trimable_id()
737        .await;
738
739    let mut scan_from_usecs = 0;
740    let mut next_full_scan = Instant::now();
741
742    loop {
743        let mut dbtx = db.begin_transaction().await;
744
745        let unordered_events = if Instant::now() >= next_full_scan {
746            next_full_scan = Instant::now() + UNORDERED_FULL_SCAN_INTERVAL;
747
748            dbtx.find_by_prefix(&UnorderedEventLogIdPrefixAll)
749                .await
750                .collect::<Vec<_>>()
751                .await
752        } else {
753            let start = UnordedEventLogId {
754                ts_usecs: scan_from_usecs,
755                counter: 0,
756            };
757
758            let end = UnordedEventLogId {
759                ts_usecs: u64::MAX,
760                counter: u64::MAX,
761            };
762
763            dbtx.find_by_range(start..end)
764                .await
765                .collect::<Vec<_>>()
766                .await
767        };
768
769        if let Some(newest_usecs) = unordered_events.iter().map(|entry| entry.0.ts_usecs).max() {
770            let margin_usecs = u64::try_from(UNORDERED_SCAN_MARGIN.as_micros()).unwrap_or(u64::MAX);
771
772            scan_from_usecs = scan_from_usecs.max(newest_usecs.saturating_sub(margin_usecs));
773        }
774        trace!(target: LOG_CLIENT_EVENT_LOG, num=unordered_events.len(), "Fetched unordered events");
775
776        for (unordered_id, entry) in &unordered_events {
777            assert!(
778                dbtx.remove_entry(unordered_id).await.is_some(),
779                "Must never fail to remove entry"
780            );
781            if entry.persist() {
782                // Non-trimable events get persisted in both the default event log
783                // and trimable event log
784                if !entry.trimable() {
785                    assert!(
786                        dbtx.insert_entry(&next_entry_id, &entry.inner)
787                            .await
788                            .is_none(),
789                        "Must never overwrite existing event"
790                    );
791                    trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Ordered event log event");
792                    next_entry_id = next_entry_id.next();
793                }
794
795                // Trimable events get persisted only in trimable log
796                assert!(
797                    dbtx.insert_entry(&next_entry_id_trimable, &entry.inner)
798                        .await
799                        .is_none(),
800                    "Must never overwrite existing event"
801                );
802                trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Ordered event log event");
803                next_entry_id_trimable = next_entry_id_trimable.next();
804            } else {
805                // Transient events don't get persisted at all
806                trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Transient event log event");
807                dbtx.on_commit({
808                    let log_event_added_transient = log_event_added_transient.clone();
809                    let entry = entry.inner.clone();
810
811                    move || {
812                        // we ignore the no-subscribers
813                        let _ = log_event_added_transient.send(entry);
814                    }
815                });
816            }
817        }
818
819        // This thread is the only thread deleting already existing element of unordered
820        // log and inserting new elements into ordered log, so it should never
821        // fail to commit.
822        dbtx.commit_tx().await;
823        if !unordered_events.is_empty() {
824            log_event_added.send_replace(());
825        }
826
827        trace!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task waits for more events");
828
829        // Waking up for the periodic full scan even without new events keeps the
830        // bound on a late commit that a seeking scan skipped. The deadline is
831        // absolute so that notifications in between cannot postpone it.
832        if let Ok(Err(_)) = timeout(
833            next_full_scan.saturating_duration_since(Instant::now()),
834            log_ordering_task_wakeup.changed(),
835        )
836        .await
837        {
838            break;
839        }
840    }
841
842    debug!(target: LOG_CLIENT_EVENT_LOG, "Event log ordering task finished");
843}
844
845/// Persistent tracker of a position in the event log
846///
847/// During processing of event log the downstream consumer needs to
848/// keep track of which event were processed already. It needs to do it
849/// atomically and persist it so event in the presence of crashes no
850/// event is ever missed or processed twice.
851///
852/// This trait allows abstracting away where and how is such position stored,
853/// e.g. which key exactly is used, in what prefixed namespace etc.
854///
855/// ## Trimmable vs Non-Trimable log
856///
857/// See [`EventPersistence`]
858#[apply(async_trait_maybe_send!)]
859pub trait EventLogNonTrimableTracker {
860    // Store position in the event log
861    async fn store(
862        &mut self,
863        dbtx: &mut DatabaseTransaction<NonCommittable>,
864        pos: EventLogId,
865    ) -> Result<(), EventLogTrackerError>;
866
867    /// Load the last previous stored position (or None if never stored)
868    async fn load(
869        &mut self,
870        dbtx: &mut DatabaseTransaction<NonCommittable>,
871    ) -> Result<Option<EventLogId>, EventLogTrackerError>;
872}
873pub type DynEventLogTracker = Box<dyn EventLogNonTrimableTracker>;
874
875/// Like [`EventLogNonTrimableTracker`] but for trimable event log
876#[apply(async_trait_maybe_send!)]
877pub trait EventLogTrimableTracker {
878    // Store position in the event log
879    async fn store(
880        &mut self,
881        dbtx: &mut DatabaseTransaction<NonCommittable>,
882        pos: EventLogTrimableId,
883    ) -> Result<(), EventLogTrackerError>;
884
885    /// Load the last previous stored position (or None if never stored)
886    async fn load(
887        &mut self,
888        dbtx: &mut DatabaseTransaction<NonCommittable>,
889    ) -> Result<Option<EventLogTrimableId>, EventLogTrackerError>;
890}
891pub type DynEventLogTrimableTracker = Box<dyn EventLogTrimableTracker>;
892
893/// A failure of an event log tracker to load or store its position in the
894/// log.
895///
896/// Trackers are implemented by applications, so the causes are their own.
897/// This type carries them unchanged: its `Display` and its `source()` are the
898/// cause's.
899#[derive(Debug, Error)]
900#[error(transparent)]
901pub struct EventLogTrackerError(Box<dyn std::error::Error + Send + Sync>);
902
903impl EventLogTrackerError {
904    /// Wraps a failure of an event log tracker, which may be any error value
905    /// or a plain message.
906    pub fn new<E>(source: E) -> Self
907    where
908        E: Into<Box<dyn std::error::Error + Send + Sync>>,
909    {
910        Self(source.into())
911    }
912}
913
914/// A failure while handling entries from the event log.
915#[derive(Debug, Error)]
916#[non_exhaustive]
917pub enum EventHandlerError<E> {
918    /// The handler rejected an event.
919    #[error("The event handler failed")]
920    Handler(#[source] E),
921
922    /// The tracker could not load or store the position in the log.
923    #[error("The event log tracker failed")]
924    Tracker(#[source] Box<dyn std::error::Error + Send + Sync>),
925}
926
927/// Feeds every entry of the event log, past and future, to `call_fn`.
928///
929/// `tracker` remembers how far the log has been read so a restart resumes where
930/// it left off; the loop stops when `call_fn` fails or the client shuts down.
931pub async fn handle_events<F, R, E>(
932    db: Database,
933    mut tracker: DynEventLogTracker,
934    mut log_event_added: watch::Receiver<()>,
935    call_fn: F,
936) -> Result<(), EventHandlerError<E>>
937where
938    F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
939    R: Future<Output = Result<(), E>>,
940    E: std::error::Error + 'static,
941{
942    let mut next_key: EventLogId = tracker
943        .load(&mut db.begin_transaction_nc().await)
944        .await
945        .map_err(|err| EventHandlerError::Tracker(err.into()))?
946        .unwrap_or_default();
947
948    trace!(target: LOG_CLIENT_EVENT_LOG, ?next_key, "Handling events");
949
950    loop {
951        let mut dbtx = db.begin_transaction().await;
952
953        match dbtx.get_value(&next_key).await {
954            Some(event) => {
955                (call_fn)(&mut dbtx.to_ref_nc(), event)
956                    .await
957                    .map_err(EventHandlerError::Handler)?;
958
959                next_key = next_key.next();
960
961                tracker
962                    .store(&mut dbtx.to_ref_nc(), next_key)
963                    .await
964                    .map_err(|err| EventHandlerError::Tracker(err.into()))?;
965
966                dbtx.commit_tx().await;
967            }
968            _ => {
969                if log_event_added.changed().await.is_err() {
970                    break Ok(());
971                }
972            }
973        }
974    }
975}
976
977/// Like [`handle_events`], for the trimable part of the log: entries are handed
978/// to `call_fn` and the tracker's position drives when they can be trimmed.
979pub async fn handle_trimable_events<F, R, E>(
980    db: Database,
981    mut tracker: DynEventLogTrimableTracker,
982    mut log_event_added: watch::Receiver<()>,
983    call_fn: F,
984) -> Result<(), EventHandlerError<E>>
985where
986    F: Fn(&mut DatabaseTransaction<NonCommittable>, EventLogEntry) -> R,
987    R: Future<Output = Result<(), E>>,
988    E: std::error::Error + 'static,
989{
990    let mut next_key: EventLogTrimableId = tracker
991        .load(&mut db.begin_transaction_nc().await)
992        .await
993        .map_err(|err| EventHandlerError::Tracker(err.into()))?
994        .unwrap_or_default();
995    trace!(target: LOG_CLIENT_EVENT_LOG, ?next_key, "Handling trimable events");
996
997    loop {
998        let mut dbtx = db.begin_transaction().await;
999
1000        match dbtx.get_value(&next_key).await {
1001            Some(event) => {
1002                (call_fn)(&mut dbtx.to_ref_nc(), event)
1003                    .await
1004                    .map_err(EventHandlerError::Handler)?;
1005
1006                next_key = next_key.next();
1007                tracker
1008                    .store(&mut dbtx.to_ref_nc(), next_key)
1009                    .await
1010                    .map_err(|err| EventHandlerError::Tracker(err.into()))?;
1011
1012                dbtx.commit_tx().await;
1013            }
1014            _ => {
1015                if log_event_added.changed().await.is_err() {
1016                    break Ok(());
1017                }
1018            }
1019        }
1020    }
1021}
1022
1023/// Filters the `PersistedLogEntries` by the `EventKind` and
1024/// `ModuleKind`.
1025pub fn filter_events_by_kind<'a, I>(
1026    all_events: I,
1027    module_kind: ModuleKind,
1028    event_kind: EventKind,
1029) -> impl Iterator<Item = &'a PersistedLogEntry> + 'a
1030where
1031    I: IntoIterator<Item = &'a PersistedLogEntry> + 'a,
1032{
1033    all_events.into_iter().filter(move |e| {
1034        if let Some(module) = &e.inner.module {
1035            e.inner.kind == event_kind && module.kind == module_kind
1036        } else {
1037            false
1038        }
1039    })
1040}
1041
1042/// Joins two sets of events on a predicate.
1043///
1044/// This function computes a "nested loop join" by first computing the cross
1045/// product of the start event vector and the success/failure event vectors. The
1046/// resulting cartesian product is then filtered according to the join predicate
1047/// supplied in the parameters.
1048///
1049/// This function is intended for small data sets. If the data set relations
1050/// grow, this function should implement a different join algorithm or be moved
1051/// out of the gateway.
1052pub fn join_events<'a, L, R, Res>(
1053    events_l: &'a [&PersistedLogEntry],
1054    events_r: &'a [&PersistedLogEntry],
1055    max_time_distance: Option<Duration>,
1056    predicate: impl Fn(L, R, Duration) -> Option<Res> + 'a,
1057) -> impl Iterator<Item = Res> + 'a
1058where
1059    L: Event,
1060    R: Event,
1061{
1062    events_l
1063        .iter()
1064        .cartesian_product(events_r)
1065        .filter_map(move |(l, r)| {
1066            if L::MODULE.as_ref() == l.as_raw().module_kind()
1067                && L::KIND == l.as_raw().kind
1068                && R::MODULE.as_ref() == r.as_raw().module_kind()
1069                && R::KIND == r.as_raw().kind
1070                && let Some(latency_usecs) = r.inner.ts_usecs.checked_sub(l.inner.ts_usecs)
1071                && max_time_distance.is_none_or(|max| u128::from(latency_usecs) <= max.as_millis())
1072                && let Some(l) = l.as_raw().to_event()
1073                && let Some(r) = r.as_raw().to_event()
1074            {
1075                predicate(l, r, Duration::from_millis(latency_usecs))
1076            } else {
1077                None
1078            }
1079        })
1080}
1081
1082/// Helper struct for storing computed data about outgoing and incoming
1083/// payments.
1084#[derive(Debug, Default)]
1085pub struct StructuredPaymentEvents {
1086    pub latencies_usecs: Vec<u64>,
1087    pub fees: Vec<Amount>,
1088    pub latencies_failure: Vec<u64>,
1089}
1090
1091impl StructuredPaymentEvents {
1092    pub fn new(
1093        success_stats: &[(u64, Amount)],
1094        failure_stats: Vec<u64>,
1095    ) -> StructuredPaymentEvents {
1096        let mut events = StructuredPaymentEvents {
1097            latencies_usecs: success_stats.iter().map(|(l, _)| *l).collect(),
1098            fees: success_stats.iter().map(|(_, f)| *f).collect(),
1099            latencies_failure: failure_stats,
1100        };
1101        events.sort();
1102        events
1103    }
1104
1105    /// Combines this `StructuredPaymentEvents` with the `other`
1106    /// `StructuredPaymentEvents` by appending all of the internal vectors.
1107    pub fn combine(&mut self, other: &mut StructuredPaymentEvents) {
1108        self.latencies_usecs.append(&mut other.latencies_usecs);
1109        self.fees.append(&mut other.fees);
1110        self.latencies_failure.append(&mut other.latencies_failure);
1111        self.sort();
1112    }
1113
1114    /// Sorts this `StructuredPaymentEvents` by sorting all of the internal
1115    /// vectors.
1116    fn sort(&mut self) {
1117        self.latencies_usecs.sort_unstable();
1118        self.fees.sort_unstable();
1119        self.latencies_failure.sort_unstable();
1120    }
1121}
1122
1123#[cfg(test)]
1124mod tests;
fedimint_eventlog/lib.rs

fedimint_eventlog/
lib.rs
