diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/access/rules.rs | 97 | ||||
| -rw-r--r-- | src/remake/catchup.rs | 1 | ||||
| -rw-r--r-- | src/remake/records.rs | 1 | ||||
| -rw-r--r-- | src/remake/snapshot.rs | 6 | ||||
| -rw-r--r-- | src/remake/switch.rs | 3 | ||||
| -rw-r--r-- | src/transport/wire.rs | 2 |
6 files changed, 86 insertions, 24 deletions
diff --git a/src/access/rules.rs b/src/access/rules.rs index 2746cb3..cf2740a 100644 --- a/src/access/rules.rs +++ b/src/access/rules.rs @@ -194,8 +194,9 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result // A node forgotten while alive cancels its own record (`swarm::peer::forget`). (Some(meta::FORGOTTEN), Some(owner)) => owner == node || user_may(path), (Some(meta::REMAKES), Some(id)) => match path.get(3).map(String::as_str) { - // Its coordinator's; or a record carried into a new epoch whole, by the - // node that coordinated it (`remake::switch`): its proposal written with it. + // Its coordinator's; or the record of the remake that made the epoch, + // carried into it whole (its proposal written with it) by the coordinator + // its snapshot names (`remake::switch`), where there is none. Some("decision") => match coordinator(&at, id) { Some(coordinator) => coordinator == node, None => { @@ -203,10 +204,11 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result .paths .iter() .any(|other| other == &meta::path(&[meta::REMAKES, id, "proposal"])) - || change - .paths - .iter() - .any(|other| other == &meta::path(&[meta::REMAKES, id])) + && store::read_path(&at, &meta::path(&[meta::REMAKES, id])) + .ok() + .flatten() + .is_none() + && made_the_epoch(&at, epoch, id, node) } }, // The record, made where there is none, and its proposal, written where @@ -410,6 +412,26 @@ fn shown(path: &Path) -> String { format!("/{}", path.join("/")) } +/// Whether remake `id`, coordinated by `node`, made `epoch`: as its snapshot's metadata +/// says (`remake::snapshot`), at `at`. Never genesis. +fn made_the_epoch(at: &At<'_>, epoch: Epoch, id: &str, node: &str) -> bool { + if epoch.generation == 0 { + return false; + } + let generation = epoch.generation.to_string(); + let Some(metadata) = store::read_path(at, &meta::path(&[meta::EPOCHS, &generation])) + .ok() + .flatten() + else { + return false; + }; + metadata.get("remakeId").and_then(serde_json::Value::as_str) == Some(id) + && metadata + .get("coordinator") + .and_then(serde_json::Value::as_str) + == Some(node) +} + /// The coordinator remake `id`'s proposal names, at `at`. fn coordinator(at: &At<'_>, id: &str) -> Option<String> { let proposal = store::read_path(at, &meta::path(&[meta::REMAKES, id, "proposal"])) @@ -544,24 +566,55 @@ mod tests { assert!(refused(judged(&doc, "b", None, &["_continuum/members"]))); assert!(refused(judged(&doc, "b", None, &["_continuum/remakes/r1"]))); assert!(judged(&doc, "b", None, &["_continuum/votes"]).is_ok()); - // A record carried into a new epoch whole: its proposal with its decision. - assert!( - judged( - &doc, - "b", - None, - &[ - "_continuum/remakes/r2", - "_continuum/remakes/r2/proposal", - "_continuum/remakes/r2/decision" - ] - ) - .is_ok() + // A record carried into a new epoch whole, its proposal with its decision: only + // the record of the remake that made the epoch, by its coordinator, where there + // is none. + let whole = [ + "_continuum/remakes/r2", + "_continuum/remakes/r2/proposal", + "_continuum/remakes/r2/decision", + ]; + assert!(refused(judged(&doc, "b", None, &whole))); + let mut later = doc.fork(); + write( + &mut later, + "_continuum/epochs/1", + json!({"generation": 1, "remakeId": "r2", "coordinator": "b"}), ); - assert!(refused(judged( - &doc, + let in_epoch_one = |doc: &Document, committer: &str, touched: &[&str]| { + let deps = doc.get_heads(); + let touched = paths(touched); + judge( + doc, + Epoch { + generation: 1, + id: [7; 32], + }, + &Change { + hash: ChangeHash([9; 32]), + deps: &deps, + seen: &deps, + committer: Some(committer), + user: None, + paths: &touched, + ops: &[], + }, + ) + }; + assert!(in_epoch_one(&later, "b", &whole).is_ok()); + assert!(refused(in_epoch_one(&later, "a", &whole))); + assert!(refused(in_epoch_one( + &later, + "b", + &[ + "_continuum/remakes/r3", + "_continuum/remakes/r3/proposal", + "_continuum/remakes/r3/decision" + ] + ))); + assert!(refused(in_epoch_one( + &later, "b", - None, &["_continuum/remakes/r2/decision"] ))); // The rest: a user's who may write there. diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs index 9830c33..4df5c90 100644 --- a/src/remake/catchup.rs +++ b/src/remake/catchup.rs @@ -718,6 +718,7 @@ mod tests { &mut old, &SnapshotParams { remake_id: "c-1", + coordinator: "a", previous: Epoch::GENESIS, base_heads: &base, policy, diff --git a/src/remake/records.rs b/src/remake/records.rs index 21b383c..21cf0ae 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -306,6 +306,7 @@ impl Record { ) -> Option<SnapshotRequest> { Some(SnapshotRequest { remake_id: id.to_owned(), + coordinator: self.proposal.coordinator.clone(), previous: self.proposal.from_epoch.to_epoch()?, base_heads: base, policy: serde_json::to_value(&self.proposal.policy).ok()?, diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs index 220989e..7938b53 100644 --- a/src/remake/snapshot.rs +++ b/src/remake/snapshot.rs @@ -36,6 +36,8 @@ use crate::document::store::{self, StoreError}; pub(crate) struct SnapshotParams<'a> { /// ID of the committed remake; seeds the snapshot actor. pub(crate) remake_id: &'a str, + /// The node coordinating it: the one that carries its record into the new epoch. + pub(crate) coordinator: &'a str, /// The epoch being compacted. The snapshot starts generation `previous + 1`. pub(crate) previous: Epoch, /// Heads of the base. Every node must hold all of these changes. @@ -84,6 +86,7 @@ pub(crate) enum SnapshotError { #[derive(Clone, Debug)] pub(crate) struct SnapshotRequest { pub(crate) remake_id: String, + pub(crate) coordinator: String, pub(crate) previous: Epoch, pub(crate) base_heads: Vec<ChangeHash>, pub(crate) policy: JsonValue, @@ -97,6 +100,7 @@ impl SnapshotRequest { pub(crate) fn params(&self) -> SnapshotParams<'_> { SnapshotParams { remake_id: &self.remake_id, + coordinator: &self.coordinator, previous: self.previous, base_heads: &self.base_heads, policy: &self.policy, @@ -311,6 +315,7 @@ pub(crate) fn build( let metadata = canonical(json!({ "generation": generation, "remakeId": params.remake_id, + "coordinator": params.coordinator, "previous": { "generation": params.previous.generation, "id": params.previous.id_hex(), @@ -391,6 +396,7 @@ mod tests { fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> { SnapshotParams { remake_id: "c-1", + coordinator: "a", previous: Epoch::GENESIS, base_heads: heads, policy, diff --git a/src/remake/switch.rs b/src/remake/switch.rs index 34c492f..bd40d72 100644 --- a/src/remake/switch.rs +++ b/src/remake/switch.rs @@ -200,7 +200,8 @@ async fn switch<H: Host>( // Only the coordinator carries the record into the new epoch: one writer, so the // nodes don't all write the same path concurrently right after switching (a merge // conflict, plus one change per node to sync). The epoch metadata in the snapshot - // keeps the essentials on every node anyway. + // keeps the essentials on every node anyway, and names the coordinator: the rules + // take the record from it only (`access::rules`). let carry_record = record.proposal.coordinator == host.node_id(); let record_path = records::record_path(id, ""); diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 7ff7835..e7bec62 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 42; +pub(crate) const VERSION: u8 = 43; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { |