simtest: forged changes, aborts of expired remakes, envelopes of another swarm

Simulations backing what the review's fixes changed, beyond their unit
tests:

* a_change_forged_under_another_members_actor (1145): a change made
  under b's actor at the seq of b's last change, judged by a: refused
  as forged, b not refused as a whole, b's next write taken.
* an_expired_remake_aborted_by_any_member (1150): under the policy, a
  coordinator gone before it decides; b aborts its remake once it
  expired a grace period ago, c takes the abort, nothing refused.
* a_strict_swarm (1155): a value alice signed for another swarm,
  replacing the one there, is refused as such.

1 files changed, 185 insertions, 0 deletions

diff --git a/src/simtest/policy.rs b/src/simtest/policy.rs
index ba32615..7f32695 100644
--- a/src/simtest/policy.rs
+++ b/src/simtest/policy.rs
@@ -361,6 +361,191 @@ async fn a_strict_swarm() {
         assert!(plain(node, "_continuum/users/mallory").await.is_none());
         assert_eq!(plain(node, "config").await, Some(json!({"mode": "slow"})));
     }
+
+    // Then a value alice signed, replacing the one there, but for another swarm: refused
+    // as such.
+    let elsewhere = {
+        use crate::access::{envelope, sshsig};
+        let previous = store::read_path(&c.doc.read().await.doc, &path("config"))
+            .ok()
+            .flatten()
+            .and_then(|value| value.as_str().map(str::to_owned));
+        let payload = envelope::Payload {
+            swarm: "elsewhere".to_owned(),
+            path: path("config"),
+            value: json!({"mode": "elsewhere"}),
+            user: "alice".to_owned(),
+            at: 2,
+            prev: envelope::digest(previous.as_deref()),
+        };
+        let bytes = serde_json::to_vec(&payload).unwrap();
+        envelope::seal(&bytes, &sshsig::sign(&key, envelope::NAMESPACE, &bytes)).unwrap()
+    };
+    forge(&c, "alice", "config", json!(elsewhere)).await;
+    sim.wait_until("the other swarm's value is refused", SOON, || async {
+        use crate::locks::MutexExt;
+        [&a, &b].iter().any(|node| {
+            node.screening
+                .lock_or_recover()
+                .refused()
+                .any(|refused| refused.why.contains("signed for swarm elsewhere"))
+        })
+    })
+    .await;
+    for node in [&a, &b] {
+        assert_eq!(plain(node, "config").await, Some(json!({"mode": "slow"})));
+    }
+}
+
+/// A remake whose coordinator is gone before it decides, under the policy: once it
+/// expired a grace period ago, any member aborts it, and every node takes the abort.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_expired_remake_aborted_by_any_member() {
+    use crate::transport::sim::LinkParams;
+    use axum::http::StatusCode;
+    let sim = Sim::with_identity(90);
+    let a = sim.start_with("a", &[], Start::Genesis, |config| {
+        config.first_users = [("alice".to_owned(), Default::default())].into();
+    });
+    let b = sim.start("b", &["a"], Start::Join);
+    let c = sim.start("c", &["a"], Start::Join);
+    sim.wait_ready(&[&a, &b, &c], SOON).await;
+    let nodes = [a.clone(), b.clone(), c.clone()];
+    sim.wait_until("every node is registered", SOON, || {
+        let nodes = nodes.clone();
+        async move {
+            for node in &nodes {
+                if crate::swarm::members::members(&node.doc.read().await.doc).len() < 3 {
+                    return false;
+                }
+            }
+            true
+        }
+    })
+    .await;
+    write_as(
+        &a,
+        "alice",
+        "_continuum/access/policy",
+        json!({"version": 1, "signatures": "required"}),
+    )
+    .await;
+    sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+    // The proposal reaches b and c, their votes never reach a, and a goes.
+    let cut = LinkParams {
+        loss: 1.0,
+        ..LinkParams::default()
+    };
+    sim.net.set_link("b", "a", cut.clone());
+    sim.net.set_link("c", "a", cut);
+    let (status, report) =
+        crate::remake::testing::start(a.clone(), json!({ "kind": "compact" }), None).await;
+    assert!(status.is_success(), "proposal refused: {status} {report}");
+    let id = report["id"].as_str().expect("remake id").to_owned();
+    let (b_ref, c_ref, id_ref) = (&b, &c, id.as_str());
+    sim.wait_until("b has the proposal", SOON, move || async move {
+        crate::remake::testing::report(b_ref, id_ref)
+            .await
+            .is_some()
+    })
+    .await;
+    sim.net.take_down("a");
+
+    sim.wait_until(
+        "b aborts the expired remake",
+        Duration::from_secs(60),
+        move || async move {
+            let (status, _) = crate::remake::testing::abort(b_ref.clone(), id_ref, None).await;
+            status == StatusCode::OK
+        },
+    )
+    .await;
+    sim.wait_until("c takes the abort", SOON, move || async move {
+        crate::remake::testing::report(c_ref, id_ref)
+            .await
+            .is_some_and(|report| report["phase"] == "aborted")
+    })
+    .await;
+    for node in [&b, &c] {
+        assert_eq!(refused(node).await, 0, "an honest change was refused");
+    }
+}
+
+/// A member making a change under another member's actor, at a seq that member's
+/// signature covers: refused as forged, by whoever brought it; the member whose actor
+/// it took isn't refused as a whole, and its next write is taken.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_change_forged_under_another_members_actor() {
+    use crate::remake::Host;
+    let sim = Sim::with_identity(91);
+    let a = sim.start_with("a", &[], Start::Genesis, |config| {
+        config.first_users = [("alice".to_owned(), Default::default())].into();
+    });
+    let b = sim.start("b", &["a"], Start::Join);
+    sim.wait_ready(&[&a, &b], SOON).await;
+    let nodes = [a.clone(), b.clone()];
+    sim.wait_until("every node is registered", SOON, || {
+        let nodes = nodes.clone();
+        async move {
+            for node in &nodes {
+                if crate::swarm::members::members(&node.doc.read().await.doc).len() < 2 {
+                    return false;
+                }
+            }
+            true
+        }
+    })
+    .await;
+    write_as(
+        &a,
+        "alice",
+        "_continuum/access/policy",
+        json!({"version": 1, "signatures": "required"}),
+    )
+    .await;
+    write_as(&b, "alice", "notes/b", json!(1)).await;
+    sim.wait_converged(&[&a, &b], SOON).await;
+
+    // On a copy of a's document: before b's last change, under b's actor, so at its seq.
+    let forged = {
+        let (actor, deps) = {
+            let replica = b.doc.read().await;
+            let last = replica.doc.get_last_local_change().expect("b wrote");
+            (replica.doc.get_actor().clone(), last.deps().to_vec())
+        };
+        let mut copy = a
+            .doc
+            .read()
+            .await
+            .doc
+            .fork_at(&deps)
+            .expect("b's change's dependencies")
+            .with_actor(actor);
+        let mut transaction = copy.transaction();
+        store::write_path(&mut transaction, &path("notes/forged"), &json!(2)).unwrap();
+        transaction.commit();
+        copy.get_last_local_change().expect("forged").clone()
+    };
+    let before = refused(&a).await;
+    a.apply_change(forged).await.expect("judged");
+    assert_eq!(refused(&a).await, before + 1);
+    assert!(!holds(&a, "notes/forged").await);
+    {
+        use crate::locks::MutexExt;
+        assert!(
+            a.screening
+                .lock_or_recover()
+                .refusing(b.local_id())
+                .is_none()
+        );
+    }
+    write_as(&b, "alice", "notes/later", json!(3)).await;
+    let a_ref = &a;
+    sim.wait_until("b's next write is taken", SOON, move || async move {
+        holds(a_ref, "notes/later").await
+    })
+    .await;
 }
 
 /// A swarm declaring values sealed (`access::sealed`), with the policy: a value sealed