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
|
//! The change publisher: after document changes, which store paths they touched
//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
//! a nudge to topology when the member list changed.
use std::sync::Arc;
use tracing::{info, warn};
use crate::document::store::ChangeSet;
use crate::state::AppState;
fn patch_store_path(patch: &automerge::Patch) -> Vec<String> {
let mut path: Vec<String> = patch
.path
.iter()
.map(|(_, prop)| match prop {
automerge::Prop::Map(key) => key.clone(),
automerge::Prop::Seq(index) => index.to_string(),
})
.collect();
match &patch.action {
automerge::PatchAction::PutMap { key, .. }
| automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()),
// List, text and counter changes: the containing object changed.
_ => {}
}
path
}
pub(crate) async fn run_change_publisher(state: AppState) {
let mut changed = state.change_tx.subscribe();
let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
info!("Starting change publisher");
while changed.changed().await.is_ok() {
let before = last_heads.clone();
let (heads, paths, conflicts) = state
.write_doc(move |doc| {
let heads = doc.get_heads();
if heads == before {
return (heads, Vec::new(), Vec::new());
}
// After an epoch switch the previous heads belong to the old document:
// report the root as changed, which every stream is related to.
if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
return (heads, vec![Vec::new()], Vec::new());
}
let mut paths = Vec::new();
let mut conflicts = Vec::new();
for patch in doc.diff(&before, &heads) {
let path = patch_store_path(&patch);
if matches!(
patch.action,
automerge::PatchAction::PutMap { conflict: true, .. }
) {
conflicts.push(path.clone());
}
paths.push(path);
}
(heads, paths, conflicts)
})
.await;
if heads == last_heads {
continue;
}
last_heads = heads;
for path in &conflicts {
warn!(path = ?path, "Concurrent writes produced a conflict");
}
if paths
.iter()
.any(|path| crate::swarm::topology::affects_neighbours(path))
{
state.topology_changed.notify_one();
}
// Fails only when nobody is subscribed.
let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
}
}
|
