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
143
144
145
146
147
148
149
150
151
152
153
154
155
|
//! 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`).
/// The node writes as a fresh actor from then on.
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;
}
// Under the write lock, and leaving the document with a fresh actor: whatever the
// node still writes until it exits (its alive stamp, a vote, a coordinator's commit:
// nothing stops them) is no change of the actor the file marks as complete.
let doc = Arc::clone(&state.doc);
let closed = tokio::task::spawn_blocking(move || {
let mut replica = doc.blocking_write();
let heads = replica.doc.get_heads();
file.close(&heads);
replica.doc.set_actor(crate::document::ActorId::random());
})
.await;
if let Err(error) = closed {
tracing::error!(error = %error, "Marking the document file as closed failed");
}
}
/// 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;
}
}
|
