diff options
| -rw-r--r-- | docs/access.md | 3 | ||||
| -rw-r--r-- | src/access/rules.rs | 14 | ||||
| -rw-r--r-- | src/access/screen.rs | 57 | ||||
| -rw-r--r-- | src/state/host.rs | 3 | ||||
| -rw-r--r-- | src/swarm/signatures.rs | 51 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 22 |
6 files changed, 98 insertions, 52 deletions
diff --git a/docs/access.md b/docs/access.md index 9d38bd5..2f36ecd 100644 --- a/docs/access.md +++ b/docs/access.md @@ -317,7 +317,8 @@ dependencies haven't come yet waits for them. A node judges what it receives whether its document has a policy yet or not, each change by the policy it was written under: a joining node judges its first sync, the -policy's own change and everything after it, as every other node does. While neither +policy's own change and everything after it, as every other node does, and waits for +the signatures of what was written under a policy requiring them. While neither its document nor what it receives holds a policy, it judges only that a root is the epoch's, and nothing costs more than applying the changes. diff --git a/src/access/rules.rs b/src/access/rules.rs index bef9a13..ac7377f 100644 --- a/src/access/rules.rs +++ b/src/access/rules.rs @@ -109,6 +109,20 @@ impl Reads { } } +impl Reads { + /// Whether the policy at `seen` requires changes signed (`Policy::required`): read + /// once for the heads, as `judge_reading` reads it. False if `doc` lacks one of them. + pub(crate) fn required(&mut self, doc: &Document, seen: &[ChangeHash]) -> bool { + let Some(at) = At::new(doc, seen) else { + return false; + }; + self.at(seen) + .policy + .get_or_insert_with(|| super::policy::of(&at)) + .required() + } +} + /// Judges `change`, about to be applied to `doc` (of `epoch`). pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result<(), Refusal> { judge_reading(doc, epoch, change, &mut Reads::default()) diff --git a/src/access/screen.rs b/src/access/screen.rs index 3a516e3..beae149 100644 --- a/src/access/screen.rs +++ b/src/access/screen.rs @@ -271,7 +271,9 @@ pub(crate) struct Screened { /// those rules allow, what they change recorded in `log`. Those already in `doc` are /// left out; those whose dependencies are missing wait; those `refusals` holds are /// refused again by their hash, and those building on them, and those refused now go -/// into it. With `signatures` (required), those no signature vouches for wait too. +/// into it. With `signatures` (the node's always; tests judge without), those written +/// under a policy requiring signatures that no signature vouches for wait too, and +/// those forged under another node's actor are refused. pub(crate) fn apply( doc: &mut Document, epoch: Epoch, @@ -437,10 +439,8 @@ pub(crate) fn apply( }); continue; } - // No signature vouches for it yet, or for what it builds on: it waits. - if signatures.is_some() && !change.deps().is_empty() && !vouched.contains(&hash) - || change.deps().iter().any(|dep| unsigned.contains(dep)) - { + // What it builds on waits for a signature: it waits too. + if change.deps().iter().any(|dep| unsigned.contains(dep)) { unsigned.insert(hash); waiting.0.push(change); continue; @@ -479,6 +479,19 @@ pub(crate) fn apply( } }; let root = change.deps().is_empty(); + // Written under a policy requiring signatures, and none vouches for it yet: it + // waits. As that policy has it, not this document's, which may have none yet + // (a joiner's) or another: so every node waits for the same. + if judging + && !root + && signatures.is_some() + && !vouched.contains(&hash) + && reads.required(doc, &seen_at) + { + unsigned.insert(hash); + waiting.0.push(change); + continue; + } let judged = if judging || root { rules::judge_reading( doc, @@ -655,8 +668,8 @@ pub(crate) struct Outcome { /// what `screening` keeps for document `epoch_seq` (of `epoch`). Bytes that don't load /// as changes, or with a misplaced author (`changes::misplaced_author`), are refused /// whole. -/// With `signatures` (required), those no signature vouches for wait too; and with no -/// bytes, those waiting are judged again (signatures came since). +/// With `signatures`, as [`apply`] has them; and with no bytes, those waiting are +/// judged again (signatures came since). pub(crate) fn apply_bytes( doc: &mut Document, epoch: Epoch, @@ -1177,6 +1190,11 @@ mod tests { }, |doc| { store::write_path(doc, &path("_continuum/swarm"), &json!("s"))?; + store::write_path( + doc, + &path("_continuum/access/policy"), + &json!({"version": 1, "signatures": "required"}), + )?; store::write_path(doc, &path("_continuum/users/alice"), &alice) }, ) @@ -1216,7 +1234,9 @@ mod tests { ) .unwrap() }; - // Unsigned: only the root goes in (no node signs it). + // Into an empty document, a joiner's, which has no policy yet: as the policy the + // changes were written under has it. Unsigned: only the root goes in (no node + // signs it). let screened = required( &mut ours, &mut waiting, @@ -1267,4 +1287,25 @@ mod tests { .is_none() ); } + + /// Written under a policy not requiring signatures: none waits for one, whatever the + /// receiving node holds. + #[test] + fn unsigned_changes_taken_where_not_required() { + let mut theirs = genesis(); + write(&mut theirs, "a", Some("alice"), "notes/x", json!(1)); + let mut ours = Document::new(); + let screened = apply( + &mut ours, + Epoch::GENESIS, + received(&theirs, &[]), + &mut Waiting::default(), + &mut Refusals::default(), + Some(&mut Signatures::default()), + &mut crate::document::PatchLog::inactive(), + ) + .unwrap(); + assert!(screened.refused.is_empty()); + assert_eq!(ours.get_heads(), theirs.get_heads()); + } } diff --git a/src/state/host.rs b/src/state/host.rs index 01cbaa2..7b2eb07 100644 --- a/src/state/host.rs +++ b/src/state/host.rs @@ -570,7 +570,6 @@ impl Host for AppState { tokio::task::spawn_blocking(move || { let result = { let mut replica = state.doc.blocking_write(); - let policy = crate::access::policy::of(&replica.doc); // Judged first (`access::screen`), whether this document has a policy // yet or not. let result = { @@ -582,7 +581,7 @@ impl Host for AppState { epoch_seq, change.raw_bytes(), &state.screening, - policy.required().then_some(&mut replica.signatures), + Some(&mut replica.signatures), &mut crate::document::PatchLog::inactive(), ) .map(|_| ()) diff --git a/src/swarm/signatures.rs b/src/swarm/signatures.rs index 9ea404e..111b3da 100644 --- a/src/swarm/signatures.rs +++ b/src/swarm/signatures.rs @@ -75,35 +75,30 @@ pub(crate) async fn received(state: &AppState, message: SignaturesMessage) { if replica.epoch != message.epoch { return 0; } - { - let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); - let replica = &mut *replica; - replica.signatures.offer(&replica.doc, signatures); - // With signatures required, changes waiting for them judged again - // (`access::screen`). - if crate::access::policy::of(&replica.doc).required() { - let applied = replica.logged_signed(|doc, known, log| { - let applied = crate::access::screen::apply_bytes( - doc, - epoch, - epoch_seq, - &[], - &screening, - Some(known), - log, - ); - (applied, true) - }); - replica.refresh_heads(); - return match applied { - Ok(outcome) => outcome.applied, - Err(error) => { - tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures"); - 0 - } - }; + let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq); + let replica = &mut *replica; + replica.signatures.offer(&replica.doc, signatures); + // Changes waiting for them judged again (`access::screen`): those the policy they + // were written under requires signed. + let applied = replica.logged_signed(|doc, known, log| { + let applied = crate::access::screen::apply_bytes( + doc, + epoch, + epoch_seq, + &[], + &screening, + Some(known), + log, + ); + (applied, true) + }); + replica.refresh_heads(); + match applied { + Ok(outcome) => outcome.applied, + Err(error) => { + tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures"); + 0 } - 0 } }) .await diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 75c20f1..e76f5f9 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -503,7 +503,7 @@ pub(crate) async fn handle_sync( let epoch_label = epoch; // Signatures of changes (`document::signatures`): a malformed list, or more than any // sender sends, counts for nothing. - let signatures = crate::document::signatures::unpack(&envelope.signatures) + let mut signatures = crate::document::signatures::unpack(&envelope.signatures) .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE) .unwrap_or_default(); let from_ready = envelope.ready; @@ -643,21 +643,18 @@ pub(crate) async fn handle_sync( let result = (|| { // Judged first, each as the policy it was written under has it // (`access::screen`), whether this document has one yet or not; - // with signatures required, by those it sent too. - let policy = crate::access::policy::of(&*doc); + // and vouched for by the signatures it sent too, where that policy + // requires them. { let bytes = message.take_changes(); - let required = policy.required(); - if required { - known.offer(doc, signatures.clone()); - } + known.offer(doc, std::mem::take(&mut signatures)); let outcome = crate::access::screen::apply_bytes( doc, epoch, epoch_seq, &bytes, &screening, - required.then_some(&mut *known), + Some(&mut *known), log, )?; refused_again = outcome.refused_again; @@ -708,10 +705,10 @@ pub(crate) async fn handle_sync( stats.asked_known_elsewhere(asked); peer_state.asked_known += (asked.0 + asked.1) as u64; replica.refresh_heads(); - // The signatures it sent, and those waiting for what it brought. - if !signatures.is_empty() || replica.heads() != heads_before.as_slice() { + // The signatures waiting for what it brought (those it sent, offered before). + if replica.heads() != heads_before.as_slice() { let replica = &mut *replica; - replica.signatures.offer(&replica.doc, signatures); + replica.signatures.offer(&replica.doc, Vec::new()); } let heads_after = replica.heads().to_vec(); if delivered { @@ -955,7 +952,6 @@ async fn download_and_apply_delta( bail!("the epoch changed during the download"); } let heads_before = replica.heads().to_vec(); - let policy = crate::access::policy::of(&replica.doc); { // Judged first (`access::screen`), whether this document has a policy yet or // not. @@ -967,7 +963,7 @@ async fn download_and_apply_delta( epoch_seq, &bytes, &screening, - policy.required().then_some(&mut replica.signatures), + Some(&mut replica.signatures), &mut crate::document::PatchLog::inactive(), )?; } |