fedimint_server/consensus/
db.rs

1use std::collections::BTreeMap;
2use std::fmt::Debug;
3
4use fedimint_core::core::ModuleInstanceId;
5use fedimint_core::db::{DatabaseTransaction, DatabaseVersion, IDatabaseTransactionOpsCoreTyped};
6use fedimint_core::encoding::{Decodable, Encodable};
7use fedimint_core::session_outcome::{AcceptedItem, ConsensusItem, SignedSessionOutcome};
8use fedimint_core::util::BoxStream;
9use fedimint_core::{
10    OutPoint, TransactionId, apply, async_trait_maybe_send, impl_db_lookup, impl_db_record,
11};
12use fedimint_server_core::migration::{
13    DynModuleHistoryItem, DynServerDbMigrationFn, IServerDbMigrationContext,
14};
15use futures::StreamExt;
16use serde::Serialize;
17
18use crate::db::DbKeyPrefix;
19
20#[derive(Clone, Debug, Encodable, Decodable)]
21pub struct AcceptedItemKey(pub u64);
22
23#[derive(Clone, Debug, Encodable, Decodable)]
24pub struct AcceptedItemPrefix;
25
26impl_db_record!(
27    key = AcceptedItemKey,
28    value = AcceptedItem,
29    db_prefix = DbKeyPrefix::AcceptedItem,
30    notify_on_modify = true,
31);
32impl_db_lookup!(key = AcceptedItemKey, query_prefix = AcceptedItemPrefix);
33
34#[derive(Debug, Encodable, Decodable, Serialize)]
35pub struct AcceptedTransactionKey(pub TransactionId);
36
37#[derive(Debug, Encodable, Decodable)]
38pub struct AcceptedTransactionKeyPrefix;
39
40impl_db_record!(
41    key = AcceptedTransactionKey,
42    value = Vec<ModuleInstanceId>,
43    db_prefix = DbKeyPrefix::AcceptedTransaction,
44    notify_on_modify = true,
45);
46impl_db_lookup!(
47    key = AcceptedTransactionKey,
48    query_prefix = AcceptedTransactionKeyPrefix
49);
50
51#[derive(Debug, Encodable, Decodable)]
52pub struct SignedSessionOutcomeKey(pub u64);
53
54#[derive(Debug, Encodable, Decodable)]
55pub struct SignedSessionOutcomePrefix;
56
57impl_db_record!(
58    key = SignedSessionOutcomeKey,
59    value = SignedSessionOutcome,
60    db_prefix = DbKeyPrefix::SignedSessionOutcome,
61    notify_on_modify = true,
62);
63impl_db_lookup!(
64    key = SignedSessionOutcomeKey,
65    query_prefix = SignedSessionOutcomePrefix
66);
67
68#[derive(Debug, Encodable, Decodable)]
69pub struct AlephUnitsKey(pub u64);
70
71#[derive(Debug, Encodable, Decodable)]
72pub struct AlephUnitsPrefix;
73
74impl_db_record!(
75    key = AlephUnitsKey,
76    value = Vec<u8>,
77    db_prefix = DbKeyPrefix::AlephUnits,
78    notify_on_modify = false,
79);
80impl_db_lookup!(key = AlephUnitsKey, query_prefix = AlephUnitsPrefix);
81
82pub fn get_global_database_migrations() -> BTreeMap<DatabaseVersion, DynServerDbMigrationFn> {
83    BTreeMap::new()
84}
85
86/// A concrete implementation of [`IServerDbMigrationContext`] APIs
87/// available for server-module db migrations.
88pub struct ServerDbMigrationContext;
89
90#[apply(async_trait_maybe_send!)]
91impl IServerDbMigrationContext for ServerDbMigrationContext {
92    async fn get_module_history_stream<'s, 'tx>(
93        &'s self,
94        module_instance_id: ModuleInstanceId,
95        dbtx: &'s mut DatabaseTransaction<'tx>,
96    ) -> BoxStream<'s, DynModuleHistoryItem>
97    where
98        'tx: 's,
99    {
100        dbtx.ensure_global().expect("Dbtx must be global");
101
102        // Items of the currently ongoing session, that have already been processed. We
103        // have to query them in full first and collect them into a vector so we don't
104        // hold two references to the dbtx at the same time.
105        let active_session_items = dbtx
106            .find_by_prefix(&AcceptedItemPrefix)
107            .await
108            .map(|(_, item)| item)
109            .collect::<Vec<_>>()
110            .await;
111
112        let stream =
113            dbtx.find_by_prefix(&SignedSessionOutcomePrefix)
114                .await
115                // Transform the session stream into an accepted item stream
116                .flat_map(|(_, signed_session_outcome): (_, SignedSessionOutcome)| {
117                    futures::stream::iter(signed_session_outcome.session_outcome.items)
118                })
119                // Append the accepted items from the current session after all the signed session
120                // items have been processed
121                .chain(futures::stream::iter(active_session_items))
122                .flat_map(move |item| {
123                    let history_items =
124                        match item.item {
125                            ConsensusItem::Transaction(tx) => {
126                                let txid = tx.tx_hash();
127                                let input_items = tx.inputs.into_iter().filter_map(|input| {
128                                    (input.module_instance_id() == module_instance_id)
129                                        .then_some(DynModuleHistoryItem::Input(input))
130                                });
131
132                                let output_items = tx.outputs.into_iter().zip(0..).filter_map(
133                                    |(output, out_idx)| {
134                                        (output.module_instance_id() == module_instance_id)
135                                            .then_some(DynModuleHistoryItem::Output(
136                                                output,
137                                                OutPoint { txid, out_idx },
138                                            ))
139                                    },
140                                );
141
142                                input_items.chain(output_items).collect::<Vec<_>>()
143                            }
144                            ConsensusItem::Module(mci) => {
145                                if mci.module_instance_id() == module_instance_id {
146                                    vec![DynModuleHistoryItem::ConsensusItem(mci)]
147                                } else {
148                                    vec![]
149                                }
150                            }
151                            ConsensusItem::Default { .. } => {
152                                unreachable!("We never save unknown CIs on the server side")
153                            }
154                        };
155                    futures::stream::iter(history_items)
156                });
157
158        Box::pin(stream)
159    }
160}
fedimint_server/consensus/db.rs

fedimint_server/consensus/
db.rs
