fedimint_server/consensus/
session_cache.rs

1//! API-only view of the latest durably completed session.
2//!
3//! Consensus and reads that must share a database snapshot must not use this
4//! cache. The completion watch publishes the index and both encodings together.
5
6use std::future::pending;
7use std::sync::Arc;
8
9use fedimint_core::db::{
10    Committable, Database, DatabaseTransaction, IDatabaseTransactionOpsCoreTyped,
11};
12use fedimint_core::module::SerdeModuleEncoding;
13use fedimint_core::session_outcome::SignedSessionOutcome;
14use futures::StreamExt;
15use tokio::sync::watch;
16
17use super::db::{SignedSessionOutcomeKey, SignedSessionOutcomePrefix};
18
19/// A completed session's canonical JSON wire values, shared without copying.
20///
21/// Caching `SerdeModuleEncoding` alone would still encode its byte vector and
22/// convert it to hex per request. Cache the result of that serialization too.
23/// The endpoint framework still allocates an owned JSON response per request.
24pub struct CachedSessionOutcome {
25    pub index: u64,
26    pub signed: Arc<serde_json::Value>,
27    pub unsigned: Arc<serde_json::Value>,
28}
29
30impl CachedSessionOutcome {
31    fn new(index: u64, outcome: &SignedSessionOutcome) -> Self {
32        Self {
33            index,
34            signed: OutcomeFormat::Signed.encode(outcome),
35            unsigned: OutcomeFormat::Unsigned.encode(outcome),
36        }
37    }
38}
39
40/// Which of the two existing outcome endpoints is being served.
41#[derive(Clone, Copy)]
42pub enum OutcomeFormat {
43    Signed,
44    Unsigned,
45}
46
47impl OutcomeFormat {
48    fn encode(self, outcome: &SignedSessionOutcome) -> Arc<serde_json::Value> {
49        Arc::new(
50            match self {
51                Self::Signed => serde_json::to_value(SerdeModuleEncoding::from(outcome)),
52                Self::Unsigned => {
53                    serde_json::to_value(SerdeModuleEncoding::from(&outcome.session_outcome))
54                }
55            }
56            .expect("Session outcome encoding is infallible"),
57        )
58    }
59
60    fn cached(self, outcome: &CachedSessionOutcome) -> Arc<serde_json::Value> {
61        match self {
62            Self::Signed => outcome.signed.clone(),
63            Self::Unsigned => outcome.unsigned.clone(),
64        }
65    }
66}
67
68pub type CompletedSession = Option<Arc<CachedSessionOutcome>>;
69
70/// Rebuild from the authoritative database before starting the API or
71/// consensus.
72pub async fn load(db: &Database) -> CompletedSession {
73    db.begin_transaction_nc()
74        .await
75        .find_by_prefix_sorted_descending(&SignedSessionOutcomePrefix)
76        .await
77        .next()
78        .await
79        .map(|(key, outcome)| Arc::new(CachedSessionOutcome::new(key.0, &outcome)))
80}
81
82/// Publish only after the transaction storing the signed outcome succeeds.
83pub fn publish_on_commit(
84    dbtx: &mut DatabaseTransaction<'_, Committable>,
85    sender: watch::Sender<CompletedSession>,
86    index: u64,
87    outcome: &SignedSessionOutcome,
88) {
89    let completed = Arc::new(CachedSessionOutcome::new(index, outcome));
90    dbtx.on_commit(move || {
91        sender.send_replace(Some(completed));
92    });
93}
94
95/// Return the exact requested session, never a newer cached session.
96pub async fn await_outcome(
97    db: &Database,
98    receiver: &watch::Receiver<CompletedSession>,
99    index: u64,
100    format: OutcomeFormat,
101) -> Arc<serde_json::Value> {
102    let mut receiver = receiver.clone();
103    loop {
104        // Mark the same snapshot we inspect as seen, before waiting. Do not hold
105        // a watch borrow across an await or retain an older outcome while waiting.
106        let latest = receiver.borrow_and_update().clone();
107        if let Some(latest) = latest {
108            if latest.index == index {
109                return format.cached(&latest);
110            }
111            if latest.index > index {
112                drop(latest);
113                let outcome = db
114                    .begin_transaction_nc()
115                    .await
116                    .get_value(&SignedSessionOutcomeKey(index))
117                    .await
118                    .expect("There are no gaps in completed session outcomes");
119                // Historical recovery is not cached: encode only the requested
120                // endpoint, and do not let old reads replace the latest session.
121                return format.encode(&outcome);
122            }
123        }
124        // The engine owns the sender for the server lifetime. At shutdown,
125        // requests are cancelled; a closed watch must not cause a busy loop.
126        if receiver.changed().await.is_err() {
127            pending::<()>().await;
128        }
129    }
130}
131
132#[cfg(test)]
133mod tests;
fedimint_server/consensus/session_cache.rs

fedimint_server/consensus/
session_cache.rs
