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

fedimint_server/consensus/
db.rs
