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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
//! 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.
//!
//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name
//! for people reading /peers, metrics and logs. Not unique and never an identity: the
//! public key stays that.
use std::collections::HashMap;
use automerge::AutoCommit;
use serde_json::{Value as JsonValue, 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
}
/// A member's stability tier, judged by itself (`swarm::stability`).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) enum Tier {
Stable,
#[default]
Transient,
}
impl Tier {
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::Stable => "stable",
Self::Transient => "transient",
}
}
}
fn tier_path(pk: &str) -> Vec<String> {
let mut path = member_path(pk);
path.push("tier".to_owned());
path
}
/// `pk`'s tier; transient unless its entry says stable.
pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier {
match store::read_path(doc, &tier_path(pk)) {
Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable,
_ => Tier::Transient,
}
}
pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> {
store::write_path(doc, &tier_path(pk), &json!(tier.as_str()))
}
fn alias_path(pk: &str) -> Vec<String> {
let mut path = member_path(pk);
path.push("alias".to_owned());
path
}
/// `pk`'s alias, if it gave itself one.
pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
match store::read_path(doc, &alias_path(pk)) {
Ok(Some(JsonValue::String(alias))) => Some(alias),
_ => None,
}
}
/// The aliases of all members that have one.
pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> {
members(doc)
.into_iter()
.filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias)))
.collect()
}
fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
match alias {
Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)),
None => match store::delete_path(doc, &alias_path(pk), false) {
Err(StoreError::NotFound) => Ok(()),
result => result,
},
}
}
/// Registers this node if it is ready and its entry is missing, and keeps its alias
/// as configured (set, changed or removed after a restart with another `--alias`).
pub(crate) async fn ensure_registered(state: &AppState) {
if !state.is_ready().await {
return;
}
let local = state.local_id().to_owned();
let own = member_path(&local);
let (registered, recorded_alias) = {
let replica = state.doc.read().await;
let registered = store::read_path(&replica.doc, &own)
.ok()
.flatten()
.is_some();
(registered, alias(&replica.doc, &local))
};
let wanted = state.alias.clone();
if !registered {
let mut entry = json!({ "since": unix_millis() });
if let Some(alias) = &wanted {
entry["alias"] = json!(alias);
}
match state
.mutate_doc(move |doc| store::write_path(doc, &own, &entry))
.await
{
Ok(()) => info!(
alias = wanted.as_deref(),
"Registered in the swarm's member list"
),
Err(error) => warn!(error = %error, "Failed to register in the member list"),
}
return;
}
if recorded_alias == wanted {
return;
}
let alias = wanted.clone();
match state
.mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref()))
.await
{
Ok(()) => info!(
alias = wanted.as_deref(),
previous = recorded_alias.as_deref(),
"Alias updated in the member list"
),
Err(error) => warn!(error = %error, "Failed to record our alias"),
}
}
/// 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,
}
}
|
