aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/config/mod.rs2
-rw-r--r--src/remake/api.rs1
-rw-r--r--src/remake/proposal.rs4
-rw-r--r--src/remake/records.rs23
-rw-r--r--src/remake/vote.rs1
-rw-r--r--src/swarm/members.rs73
6 files changed, 94 insertions, 10 deletions
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 0101aa4..43dd408 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -254,7 +254,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
Kind::Duration,
"DURATION",
Some("7d"),
- "Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). At least 7h (members renew every seventh of their own window, hourly at most); the same on every node, or a coordinator with a shorter one counts fewer members.",
+ "Members that renewed their alive stamp within this time vote on remakes (a majority of them must agree). At least 7h. Members renew their stamps every seventh of the shortest window any of them publishes, hourly at most, so nodes may differ. Matters on the node that coordinates a remake.",
),
option(
"superseded-writes",
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 7750242..8fd0c5b 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -453,6 +453,7 @@ async fn start_kind<H: Host>(
proposed_at: now,
settle_until: now.plus(settle),
switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64),
+ grace_ms: Some(grace),
// Filled in when the proposal is written, from the document at that moment.
electorate: Vec::new(),
quorum: 0,
diff --git a/src/remake/proposal.rs b/src/remake/proposal.rs
index d9cd893..7695b4c 100644
--- a/src/remake/proposal.rs
+++ b/src/remake/proposal.rs
@@ -23,7 +23,7 @@ pub(super) fn abortable_at(record: &Record, grace: u64) -> UnixMillis {
record
.proposal
.settle_until
- .plus_ms(grace.saturating_mul(2))
+ .plus_ms(record.grace(grace).saturating_mul(2))
}
/// Why this node may not abort expired `record` now, and the `force` name that would
@@ -157,7 +157,7 @@ pub(super) async fn propose<H: Host>(
coordinator_active: supersede
.active_peers
.contains(&record.proposal.coordinator),
- since: record.proposal.settle_until.plus_ms(grace),
+ since: record.proposal.settle_until.plus_ms(record.grace(grace)),
abortable_at: abortable_at(record, grace),
tally: vote::tally(id, record, &all, &register),
})
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 55fa8a0..0c4c444 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -122,6 +122,11 @@ pub(crate) struct Proposal {
/// interval if absent.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) switch_delay_ms: Option<u64>,
+ /// The coordinator's grace period (ms, its `dormant_after`): every node judges this
+ /// remake's expiry, and when others may abort it, by it, whatever its own keepalive
+ /// (`Record::grace`). Absent in records from before: each node's own.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub(crate) grace_ms: Option<u64>,
/// Members whose votes count (sorted), fixed at proposal time (`vote`).
pub(crate) electorate: Vec<String>,
/// Votes needed to commit: a majority of the electorate.
@@ -311,14 +316,23 @@ impl Record {
})
}
- pub(crate) fn phase(&self, now: UnixMillis, grace_ms: u64) -> Phase {
+ /// The phase at `now`, expiry judged by the record's grace period (`grace`; `own`,
+ /// this node's, for a record without one).
+ pub(crate) fn phase(&self, now: UnixMillis, own: u64) -> Phase {
match &self.decision {
Some(Decision::Committed { .. }) => Phase::Committed,
Some(Decision::Aborted { .. }) => Phase::Aborted,
- None if now > self.proposal.settle_until.plus_ms(grace_ms) => Phase::Expired,
+ None if now > self.proposal.settle_until.plus_ms(self.grace(own)) => Phase::Expired,
None => Phase::Proposed,
}
}
+
+ /// The grace period this remake is judged by (ms): its coordinator's, the same on
+ /// every node, so nodes with different keepalives agree on when it expired; `own`
+ /// for a record from before it was written down.
+ pub(crate) fn grace(&self, own: u64) -> u64 {
+ self.proposal.grace_ms.unwrap_or(own)
+ }
}
// --- document access ---------------------------------------------------------
@@ -564,6 +578,7 @@ mod tests {
proposed_at: UnixMillis(settle_until.saturating_sub(1_000)),
settle_until: UnixMillis(settle_until),
switch_delay_ms: None,
+ grace_ms: None,
electorate: vec!["node-a".into(), "node-b".into()],
quorum: 2,
kind: DEFAULT_KIND.to_owned(),
@@ -654,6 +669,10 @@ mod tests {
assert_eq!(record.phase(UnixMillis(10_000), 500), Phase::Proposed);
assert_eq!(record.phase(UnixMillis(10_500), 500), Phase::Proposed);
assert_eq!(record.phase(UnixMillis(10_501), 500), Phase::Expired);
+ // The coordinator's grace period, written down, wins over this node's.
+ record.proposal.grace_ms = Some(2_000);
+ assert_eq!(record.phase(UnixMillis(10_501), 500), Phase::Proposed);
+ assert_eq!(record.phase(UnixMillis(12_001), 500), Phase::Expired);
record.decision = Some(Decision::Aborted {
reason: "test".into(),
decided_at: UnixMillis(1),
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index 6604ff6..12c1333 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -389,6 +389,7 @@ mod tests {
proposed_at,
settle_until: proposed_at.plus_ms(1_000),
switch_delay_ms: None,
+ grace_ms: None,
electorate: electorate.iter().map(|m| (*m).to_owned()).collect(),
quorum: quorum(electorate.len()),
kind: "compact".to_owned(),
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 098c767..59c7571 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -225,9 +225,27 @@ pub(crate) fn electorate(doc: &impl ReadDoc, now_ms: u64, window: Duration) -> V
.collect()
}
+fn window_path(pk: &str) -> Vec<String> {
+ let mut path = member_path(pk);
+ path.push("window".to_owned());
+ path
+}
+
+/// The shortest `--electorate-window` among this node's (`own`) and those the members
+/// publish in their entries (ms, `window`): every node renews its alive stamp by it
+/// (`alive_renewal`), so a coordinator with a shorter window than others' still counts
+/// them all. Below `MIN_ELECTORATE_WINDOW` (which no node accepts) counts as that.
+pub(crate) fn shortest_window(doc: &impl ReadDoc, own: Duration) -> Duration {
+ members(doc)
+ .iter()
+ .filter_map(|pk| millis_at(doc, &window_path(pk)))
+ .map(|ms| Duration::from_millis(ms).max(MIN_ELECTORATE_WINDOW))
+ .fold(own, Duration::min)
+}
+
/// How often a node renews its `alive` stamp: often enough that a live node never
-/// drops out of the electorate (a seventh of the window), but at most hourly and at
-/// least daily, so idle swarms stay quiet.
+/// drops out of the electorate (a seventh of the shortest window, `shortest_window`),
+/// but at most hourly and at least daily, so idle swarms stay quiet.
fn alive_renewal(window: Duration) -> Duration {
(window / 7).clamp(RENEWAL_MOST_OFTEN, Duration::from_secs(24 * 3600))
}
@@ -249,7 +267,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
}
let local = state.local_id().to_owned();
let own = member_path(&local);
- let (registered, recorded_alias, recorded_addresses, alive) = {
+ let window = state.electorate_window.as_millis() as u64;
+ let (registered, recorded_alias, recorded_addresses, alive, recorded_window, shortest) = {
let replica = state.doc.read().await;
let registered = store::read_path(&replica.doc, &own)
.ok()
@@ -260,6 +279,8 @@ pub(crate) async fn ensure_registered(state: &AppState) {
alias(&replica.doc, &local),
addresses(&replica.doc, &local),
millis_at(&replica.doc, &alive_path(&local)),
+ millis_at(&replica.doc, &window_path(&local)),
+ shortest_window(&replica.doc, state.electorate_window),
)
};
let wanted = state.alias.clone();
@@ -268,7 +289,12 @@ pub(crate) async fn ensure_registered(state: &AppState) {
let now = state.clock.now_millis().as_u64();
if !registered {
- let mut entry = json!({ "since": now, "alive": now, "addresses": own_addresses });
+ let mut entry = json!({
+ "since": now,
+ "alive": now,
+ "addresses": own_addresses,
+ "window": window,
+ });
if let Some(alias) = &wanted {
entry["alias"] = json!(alias);
}
@@ -285,7 +311,21 @@ pub(crate) async fn ensure_registered(state: &AppState) {
return;
}
- let renewal = alive_renewal(state.electorate_window).as_millis() as u64;
+ if recorded_window != Some(window) {
+ let path = window_path(&local);
+ match state
+ .mutate_doc(move |doc| store::write_path(doc, &path, &json!(window)))
+ .await
+ {
+ Ok(()) => debug!(
+ window_ms = window,
+ "Electorate window published in the member list"
+ ),
+ Err(error) => warn!(error = %error, "Failed to record our electorate window"),
+ }
+ }
+
+ let renewal = alive_renewal(shortest).as_millis() as u64;
if alive.is_none_or(|at| now.saturating_sub(at) >= renewal) {
let path = alive_path(&local);
match state
@@ -377,6 +417,29 @@ mod tests {
}
#[test]
+ fn shortest_published_window() {
+ let day = Duration::from_secs(24 * 3600);
+ let mut doc = Document::new();
+ assert_eq!(shortest_window(&doc, 7 * day), 7 * day);
+ change(&mut doc, |doc| {
+ store::write_path(
+ doc,
+ &member_path("a"),
+ &json!({ "window": 30 * 86_400_000_u64 }),
+ )
+ .unwrap();
+ store::write_path(doc, &member_path("b"), &json!({ "window": 86_400_000 })).unwrap();
+ store::write_path(doc, &member_path("c"), &json!({ "since": 1 })).unwrap();
+ });
+ assert_eq!(shortest_window(&doc, 7 * day), day);
+ // Below what any node accepts: as the shortest it accepts.
+ change(&mut doc, |doc| {
+ store::write_path(doc, &window_path("c"), &json!(1)).unwrap();
+ });
+ assert_eq!(shortest_window(&doc, 7 * day), MIN_ELECTORATE_WINDOW);
+ }
+
+ #[test]
fn renewal_interval() {
let day = Duration::from_secs(24 * 3600);
assert_eq!(alive_renewal(7 * day), day);