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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
//! The worker that keeps the document file (`document::file`) up to date: it looks at
//! the node (ready, the document's epoch and heads, this node's actors), and the file
//! says what is due. While a ready node rejoins, the file follows the document before
//! the rejoin, which takes the node's writes meanwhile (`state::StaleView`): a restart
//! then resumes on it, and rejoins again.
use std::sync::Arc;
use crate::document::Document;
use crate::document::file::Due;
use crate::document::replica::Epoch;
use crate::locks::RwLockExt;
use crate::state::AppState;
/// Keeps the document file up to date. Runs for as long as the node does.
pub(crate) async fn run(state: AppState) {
let Some(file) = state.document_file.clone() else {
return;
};
let mut tick = tokio::time::interval(file.append_every());
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tick.tick().await;
flush(&state).await;
}
}
/// At a clean stop, after the last `flush`: the file marked as holding everything this
/// node wrote (`DocumentFile::close`), where it does (ready, not rejoining, up to the
/// document's heads), so a restart may write on as the same actor (`--keep-actor`).
pub(crate) async fn close(state: &AppState) {
let Some(file) = state.document_file.clone() else {
return;
};
if state.remake.is_rejoining() || !state.is_ready().await {
return;
}
let heads = state.doc.read().await.doc.get_heads();
file.close(&heads);
}
/// Writes what is due now (`DocumentFile::due`). Also called at shutdown.
pub(crate) async fn flush(state: &AppState) {
let Some(file) = state.document_file.clone() else {
return;
};
if state.remake.is_rejoining() {
if let Some((epoch, stale)) = state.stale_document() {
flush_before_rejoin(&file, epoch, stale).await;
}
return;
}
if !state.is_ready().await {
return;
}
let own = state.remake.own_actors();
let (epoch, heads, signatures) = {
let replica = state.doc.read().await;
(
replica.epoch,
replica.heads().to_vec(),
replica.signatures.count(),
)
};
match file.due(epoch, own.len(), &heads, Some(signatures)) {
Due::Rewrite(reason) => {
let (epoch, copy, signatures) = {
let replica = state.doc.read().await;
(
replica.epoch,
replica.doc.clone(),
replica.signatures.since(0),
)
};
file.rewrite(epoch, copy, signatures, own, reason).await;
}
Due::Append {
epoch,
since,
signatures,
} => {
let (heads, increment, signatures) = {
let replica = state.doc.read().await;
// Switched since: the next flush rewrites the file instead.
if replica.epoch != epoch {
return;
}
let heads = replica.heads().to_vec();
(
heads,
crate::document::save_since(&replica.doc, &since),
replica.signatures.since(signatures),
)
};
file.append(since, heads, increment, signatures).await;
}
Due::Nothing => {}
}
}
/// [`flush`] of the document before a rejoin (`epoch`'s), while it takes this node's
/// writes. Only appends: a rewrite (another epoch, or the increments outgrew the full
/// save) waits for the end of the rejoin, so the file keeps the actors it had.
async fn flush_before_rejoin(
file: &Arc<crate::document::file::DocumentFile>,
epoch: Epoch,
stale: Arc<std::sync::RwLock<Document>>,
) {
let heads = {
let stale = Arc::clone(&stale);
// Sorted, as the live document's (`Replica::heads`) the file compares with.
tokio::task::spawn_blocking(move || {
let mut heads = stale.read_or_recover().get_heads();
heads.sort_unstable();
heads
})
.await
.expect("document read task panicked")
};
let Some(actors) = file.actors() else {
return;
};
// Its signatures aren't kept: those of the document after the rejoin are.
if let Due::Append {
epoch: written,
since,
..
} = file.due(epoch, actors, &heads, None)
&& written == epoch
{
let since_then = since.clone();
let (heads, increment) = tokio::task::spawn_blocking(move || {
let doc = stale.read_or_recover();
let mut heads = doc.get_heads();
heads.sort_unstable();
(heads, crate::document::save_since(&doc, &since_then))
})
.await
.expect("document read task panicked");
file.append(since, heads, increment, Vec::new()).await;
}
}
|
