1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
//! Serving reads: from the live document, or from the pre-rejoin document while
//! rejoining.
use std::sync::Arc;
use super::AppState;
use crate::document::replica::Epoch;
use crate::document::{ChangeHash, Document};
use crate::locks::RwLockExt;
/// The document the node served before it started rejoining, of the epoch it left:
/// reads are answered from it (marked stale), and writes go into it
/// (`AppState::mutate_doc`), until the rejoin replays them into the new epoch or
/// restores this document (`Host::finish_rejoin`, `Host::abandon_rejoin`).
pub(crate) struct StaleView {
pub(crate) epoch: Epoch,
pub(crate) doc: Arc<std::sync::RwLock<Document>>,
}
/// Where a read was served from.
pub(crate) enum ReadSource {
Live,
/// The pre-rejoin document of this epoch.
Stale(Epoch),
}
impl AppState {
/// Renders a read for the store API: from the live document when the node is ready,
/// from the stale view while it rejoins, and otherwise `503`, or the live document
/// anyway with `unready_live`.
///
/// With `with_heads`, a live read also returns the document's heads at the moment of
/// the read (for clients that base later requests on what they read). Stale reads
/// have no heads.
pub(crate) async fn render_read<T, R>(
&self,
unready_live: bool,
with_heads: bool,
read: impl FnOnce(&Document) -> T + Send + 'static,
finish: impl FnOnce(T) -> R + Send + 'static,
) -> Result<(R, ReadSource, Option<Vec<ChangeHash>>), axum::http::StatusCode>
where
T: Send + 'static,
R: Send + 'static,
{
if self.is_ready().await {
let (rendered, heads) = self.render_live(with_heads, read, finish).await;
return Ok((rendered, ReadSource::Live, heads));
}
let stale = self
.stale
.read_or_recover()
.as_ref()
.map(|view| (view.epoch, Arc::clone(&view.doc)));
let Some((epoch, doc)) = stale else {
if unready_live {
let (rendered, heads) = self.render_live(with_heads, read, finish).await;
return Ok((rendered, ReadSource::Live, heads));
}
return Err(axum::http::StatusCode::SERVICE_UNAVAILABLE);
};
let rendered = tokio::task::spawn_blocking(move || {
let data = read(&doc.read_or_recover());
finish(data)
})
.await
.expect("document read task panicked");
Ok((rendered, ReadSource::Stale(epoch), None))
}
async fn render_live<T, R>(
&self,
with_heads: bool,
read: impl FnOnce(&Document) -> T + Send + 'static,
finish: impl FnOnce(T) -> R + Send + 'static,
) -> (R, Option<Vec<ChangeHash>>)
where
T: Send + 'static,
R: Send + 'static,
{
if !with_heads {
return (self.render(read, finish).await, None);
}
// Heads and data under the same read lock (the heads are cached: `Replica`).
let doc = Arc::clone(&self.doc);
let (data, heads) = tokio::task::spawn_blocking(move || {
let replica = doc.blocking_read();
(read(&replica.doc), replica.heads().to_vec())
})
.await
.expect("document read task panicked");
let rendered = tokio::task::spawn_blocking(move || finish(data))
.await
.expect("document read task panicked");
(rendered, Some(heads))
}
pub(super) fn set_stale(&self, view: Option<StaleView>) {
*self.stale.write_or_recover() = view;
}
/// The stale view's document and epoch, if there is one.
pub(crate) fn stale_document(&self) -> Option<(Epoch, Arc<std::sync::RwLock<Document>>)> {
self.stale
.read_or_recover()
.as_ref()
.map(|view| (view.epoch, Arc::clone(&view.doc)))
}
}
|
