aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/access/rules.rs14
-rw-r--r--src/access/screen.rs57
-rw-r--r--src/state/host.rs3
-rw-r--r--src/swarm/signatures.rs51
-rw-r--r--src/swarm/sync.rs22
5 files changed, 96 insertions, 51 deletions
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(),
)?;
}