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
|
//! The member registry: every node that has joined the swarm, under
//! `_continuum/members/<pk>`, replicated with the document (and carried into new
//! epochs with the rest of `_continuum`).
//!
//! A pure registry: it decides nothing about admission (that's the swarm key). Each
//! node registers itself once it is ready, and re-registers if its entry disappears
//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry
//! together with writing the forget record; a forgotten node that turns out to be alive
//! cancels the record and registers again.
use automerge::AutoCommit;
use serde_json::json;
use tracing::{info, warn};
use crate::document::store::{self, StoreError};
use crate::state::{AppState, unix_millis};
const META_ROOT: &str = "_continuum";
fn members_dir() -> Vec<String> {
vec![META_ROOT.to_owned(), "members".to_owned()]
}
pub(crate) fn member_path(pk: &str) -> Vec<String> {
let mut path = members_dir();
path.push(pk.to_owned());
path
}
/// Public keys of all registered members, sorted.
pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
members.sort();
members
}
/// Registers this node if it is ready and its entry is missing.
pub(crate) async fn ensure_registered(state: &AppState) {
if !state.is_ready().await {
return;
}
let own = member_path(state.local_id());
let registered = store::read_path(&state.doc.read().await.doc, &own)
.ok()
.flatten()
.is_some();
if registered {
return;
}
let entry = json!({ "since": unix_millis() });
match state
.mutate_doc(move |doc| store::write_path(doc, &own, &entry))
.await
{
Ok(()) => info!("Registered in the swarm's member list"),
Err(error) => warn!(error = %error, "Failed to register in the member list"),
}
}
/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
match store::delete_path(doc, &member_path(pk), false) {
Err(StoreError::NotFound) => Ok(()),
result => result,
}
}
|
