remake: aborts written as the user who asks; no abort by time

Review 2, N3: 1150 let any member write an abort decision once a remake
had expired a grace period ago, judged by the change's own time, the
writer's word. A member could race the coordinator's legitimate commit
with an abort dated as it liked; both taken, Automerge shows
one, and nodes that acted on the commit split from those that didn't.

The two writers of such aborts are operator requests whose endpoints
already need write on _continuum/remakes: aborting an expired remake
(DELETE /admin/remakes/{id}) and superseding one (POST
/admin/remakes?force=supersede). Both now commit as the user who asks
(the proposal and its vote with them, for a supersede), which the rules
take as one who may write the decision. The clause judging an abort by
its time goes, and with it the time the rules carried. An expired
remake is still aborted through any member, by an operator.

7 files changed, 69 insertions, 180 deletions

diff --git a/docs/access.md b/docs/access.md
index d7e05ab..4400b91 100644
--- a/docs/access.md
+++ b/docs/access.md
@@ -245,12 +245,11 @@ stays as it is, wherever it comes from later (a joining node catching up, say).
   change only (the genesis change, a remake's snapshot).
 * A remake's record and its proposal by whichever member proposes, where there are
   none; its decision by the coordinator its proposal names, or a user who may write
-  there; an abort by any member once the remake expired a grace period ago, by the
-  time the change says it was made (its writer's word: a member could abort a remake
-  early, never commit one); an abort asked for by any member, which the coordinator
-  applies. The record of the remake that made an epoch, carried into it whole, by the
-  coordinator its snapshot names, where there is none. Replacing or removing a record or its proposal, by a user who may write
-  there.
+  there (an operator aborting an expired remake through any member, `DELETE
+  /admin/remakes/{id}`, or superseding it); an abort asked for by any member, which the
+  coordinator applies. The record of the remake that made an epoch, carried into it
+  whole, by the coordinator its snapshot names, where there is none. Replacing or
+  removing a record or its proposal, by a user who may write there.
 * What holds the members' entries, votes and remake records (`_continuum/members`,
   `_continuum/votes`, `_continuum/remakes`) made by whichever member first writes in
   it, where there is none; replaced or removed by a user who may write there only.
diff --git a/docs/remakes.md b/docs/remakes.md
index fc938c7..ab808e5 100644
--- a/docs/remakes.md
+++ b/docs/remakes.md
@@ -255,8 +255,9 @@ aborts it itself, under three conditions that keep it from crossing a late commi
   Clocks): whether the remake expired is judged by it, and a node ahead would abort a
   remake its coordinator may still commit. Otherwise `409`, unless `?force=clock`.
 
-With a policy ([`access.md`](access.md)), the other nodes take that abort by the time
-it says it was written: a grace period after the expiry at least.
+The abort is written as the user who asks, who may write remakes: with a policy
+([`access.md`](access.md)), the other nodes take a decision from the coordinator or
+from such a user only.
 
 A member that saw a remake expire before it voted for it says so in its log, once
 (`Didn't vote for a remake`, warn if its clock is off: it then misses remakes others
diff --git a/src/access/rules.rs b/src/access/rules.rs
index b5dcf3a..ed9ee4b 100644
--- a/src/access/rules.rs
+++ b/src/access/rules.rs
@@ -56,8 +56,6 @@ pub(crate) struct Change<'a> {
     pub(crate) paths: &'a BTreeSet<Path>,
     /// Its operations, where each is and what it does there (`inspect::Inspector::ops`).
     pub(crate) ops: &'a [Op],
-    /// When it was made (ms), as its writer says.
-    pub(crate) time: u64,
 }
 
 /// Why a change is refused.
@@ -214,12 +212,10 @@ pub(crate) fn judge(doc: &Document, epoch: Epoch, change: &Change<'_>) -> Result
                 // 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.
-                // Also by one who may write it; and an abort by any member, once the
-                // remake expired a grace period ago (`aborts_expired`).
+                // Also by one who may write it (an operator aborting an expired remake,
+                // or superseding it, `remake::api`).
                 Some("decision") => match coordinator(&at, id) {
-                    Some(coordinator) => {
-                        coordinator == node || user_may(path) || aborts_expired(&at, id, change)
-                    }
+                    Some(coordinator) => coordinator == node || user_may(path),
                     None if user_may(path) => true,
                     None => {
                         change
@@ -306,7 +302,6 @@ pub(crate) fn judge_held(
             user: user.as_deref(),
             paths: &paths,
             ops: &ops,
-            time: u64::try_from(change.timestamp()).unwrap_or(0),
         },
     );
     Some(match judged {
@@ -455,38 +450,6 @@ fn made_the_epoch(at: &At<'_>, epoch: Epoch, id: &str, node: &str) -> bool {
             == Some(node)
 }
 
-/// Whether `change` aborts remake `id`, undecided at `at`, as any member may: once it
-/// expired a grace period ago (`remake::proposal::abortable_at`, its coordinator's grace
-/// period), by the change's time. That is its writer's word: a member could abort a
-/// remake early, as it could ask its coordinator to; never commit one.
-fn aborts_expired(at: &At<'_>, id: &str, change: &Change<'_>) -> bool {
-    if store::read_path(at, &meta::path(&[meta::REMAKES, id, "decision"]))
-        .ok()
-        .flatten()
-        .is_some()
-    {
-        return false;
-    }
-    let outcome = meta::path(&[meta::REMAKES, id, "decision", "outcome"]);
-    let mut outcomes = change.ops.iter().filter(|op| op.path == outcome).peekable();
-    if outcomes.peek().is_none()
-        || !outcomes.all(|op| matches!(&op.action, Action::PutString(value) if value == "aborted"))
-    {
-        return false;
-    }
-    let Some(proposal) = store::read_path(at, &meta::path(&[meta::REMAKES, id, "proposal"]))
-        .ok()
-        .flatten()
-    else {
-        return false;
-    };
-    let number = |key: &str| proposal.get(key).and_then(serde_json::Value::as_u64);
-    let (Some(settle_until), Some(grace)) = (number("settleUntil"), number("graceMs")) else {
-        return false;
-    };
-    change.time >= settle_until.saturating_add(grace.saturating_mul(2))
-}
-
 /// Whether `owner` has a key left after `ops`, its keys at `at` as they change them.
 fn keys_left(at: &At<'_>, owner: &str, ops: &[Op]) -> bool {
     let keys = meta::path(&[meta::USERS, owner, "keys"]);
@@ -593,7 +556,6 @@ mod tests {
                 user,
                 paths: &touched,
                 ops: &[],
-                time: 0,
             },
         )
     }
@@ -678,7 +640,6 @@ mod tests {
                     user: None,
                     paths: &touched,
                     ops: &[],
-                    time: 0,
                 },
             )
         };
@@ -766,130 +727,26 @@ mod tests {
                 user: None,
                 paths: &touched,
                 ops: &[],
-                time: 0,
             },
         );
         assert!(earlier.is_ok());
     }
 
-    /// A remake's decision: its coordinator's, one who may write it; an abort, any
-    /// member's, once the remake expired a grace period ago, by the change's time.
+    /// A remake's decision: its coordinator's, or one who may write it; no other
+    /// member's, whatever its time.
     #[test]
-    fn expired_remakes_aborted_by_any_member() {
+    fn decisions_by_the_coordinator_or_who_may() {
         let mut doc = swarm(true);
         write(
             &mut doc,
-            "_continuum/remakes/r4/proposal",
-            json!({"coordinator": "a", "settleUntil": 1_000, "graceMs": 100}),
-        );
-        write(
-            &mut doc,
             "_continuum/users/carol",
             registry::entry("a", 2, &[(path("_continuum"), &[Right::Write][..])]),
         );
-        let decide = |user: Option<&str>, outcome: &str, time: u64| {
-            let ops = vec![
-                Op {
-                    path: path("_continuum/remakes/r4/decision"),
-                    action: Action::MakeMap,
-                },
-                Op {
-                    path: path("_continuum/remakes/r4/decision/outcome"),
-                    action: Action::PutString(outcome.to_owned()),
-                },
-                Op {
-                    path: path("_continuum/remakes/r4/decision/decidedAt"),
-                    action: Action::Other,
-                },
-            ];
-            let deps = doc.get_heads();
-            let touched: BTreeSet<Path> = ops.iter().map(|op| op.path.clone()).collect();
-            judge(
-                &doc,
-                Epoch::GENESIS,
-                &Change {
-                    hash: ChangeHash([9; 32]),
-                    deps: &deps,
-                    seen: &deps,
-                    committer: Some("b"),
-                    user,
-                    paths: &touched,
-                    ops: &ops,
-                    time,
-                },
-            )
-        };
-        // Expired at 1100; abortable by others a grace period later.
-        assert!(refused(decide(None, "aborted", 1_199)));
-        assert!(decide(None, "aborted", 1_200).is_ok());
-        assert!(refused(decide(None, "committed", 5_000)));
-        // One who may write protocol data, any decision, any time.
-        assert!(decide(Some("carol"), "committed", 0).is_ok());
-    }
-
-    /// A user's own keys: its own to change, whatever else it may do, one left at least;
-    /// another's, or the rest of its own entry, not.
-    #[test]
-    fn users_change_their_own_keys() {
-        let mut doc = swarm(true);
-        let judged_ops = |doc: &Document, user: Option<&str>, ops: Vec<Op>| {
-            let deps = doc.get_heads();
-            let touched: BTreeSet<Path> = ops.iter().map(|op| op.path.clone()).collect();
-            judge(
-                doc,
-                Epoch::GENESIS,
-                &Change {
-                    hash: ChangeHash([9; 32]),
-                    deps: &deps,
-                    seen: &deps,
-                    committer: Some("a"),
-                    user,
-                    paths: &touched,
-                    ops: &ops,
-                    time: 0,
-                },
-            )
-        };
-        let op = |at: &str, action: Action| Op {
-            path: path(at),
-            action,
-        };
-        let put = |at: &str| op(at, Action::PutString("ssh-ed25519 AAAA".to_owned()));
-        let bobs = "_continuum/users/bob/keys/laptop";
-        assert!(judged_ops(&doc, Some("bob"), vec![put(bobs)]).is_ok());
-        assert!(refused(judged_ops(
-            &doc,
-            Some("bob"),
-            vec![put("_continuum/users/alice/keys/laptop")]
-        )));
-        assert!(refused(judged_ops(
-            &doc,
-            Some("bob"),
-            vec![op("_continuum/users/bob/caps", Action::MakeMap)]
-        )));
-        assert!(refused(judged_ops(&doc, None, vec![put(bobs)])));
-        // With one key: another added, one of two removed; not the last.
-        write(&mut doc, bobs, json!("ssh-ed25519 AAAA"));
-        let phone = "_continuum/users/bob/keys/phone";
-        assert!(judged_ops(&doc, Some("bob"), vec![put(phone)]).is_ok());
-        assert!(refused(judged_ops(
-            &doc,
-            Some("bob"),
-            vec![op(bobs, Action::Delete)]
-        )));
-        assert!(
-            judged_ops(
-                &doc,
-                Some("bob"),
-                vec![put(phone), op(bobs, Action::Delete)]
-            )
-            .is_ok()
-        );
-        assert!(refused(judged_ops(
-            &doc,
-            Some("bob"),
-            vec![op("_continuum/users/bob/keys", Action::Delete)]
-        )));
+        let decision = "_continuum/remakes/r1/decision";
+        assert!(judged(&doc, "a", None, &[decision]).is_ok());
+        assert!(refused(judged(&doc, "b", None, &[decision])));
+        assert!(refused(judged(&doc, "b", Some("bob"), &[decision])));
+        assert!(judged(&doc, "b", Some("carol"), &[decision]).is_ok());
     }
 
     /// Under a strict policy, a user's own entry is its own to write (what it may sign
@@ -973,7 +830,6 @@ mod tests {
                     user: Some("alice"),
                     paths: &touched,
                     ops: &[],
-                    time: 0,
                 },
             )
         };
@@ -1037,7 +893,6 @@ mod tests {
                     user: Some("alice"),
                     paths: &touched,
                     ops: &ops,
-                    time: 0,
                 },
             )
         };
@@ -1128,7 +983,6 @@ mod tests {
                     user: Some("alice"),
                     paths: &touched,
                     ops: &ops,
-                    time: 0,
                 },
             )
         };
@@ -1206,7 +1060,6 @@ mod tests {
                     user: Some("bob"),
                     paths: &touched,
                     ops: &ops,
-                    time: 0,
                 },
             )
         };
@@ -1261,7 +1114,6 @@ mod tests {
                     user: None,
                     paths: &touched,
                     ops: &[],
-                    time: 0,
                 },
             )
         };
diff --git a/src/access/screen.rs b/src/access/screen.rs
index 5584b2b..d7c9389 100644
--- a/src/access/screen.rs
+++ b/src/access/screen.rs
@@ -442,7 +442,6 @@ pub(crate) fn apply(
                 user: user.as_deref(),
                 paths: &paths[index],
                 ops: &ops[index],
-                time: u64::try_from(change.timestamp()).unwrap_or(0),
             },
         );
         match judged {
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 7aeb675..d3df908 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -302,11 +302,12 @@ async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
 /// `POST /admin/remakes`.
 pub(crate) async fn start<H: Host>(
     State(host): State<H>,
+    axum::Extension(user): axum::Extension<crate::api::auth::AuthenticatedUser>,
     Query(query): Query<StartQuery>,
     headers: HeaderMap,
     body: Bytes,
 ) -> Response {
-    start_kind(host, query, &headers, body).await
+    start_kind(host, query, &headers, body, Some(user.0)).await
 }
 
 async fn start_kind<H: Host>(
@@ -314,6 +315,7 @@ async fn start_kind<H: Host>(
     query: StartQuery,
     headers: &HeaderMap,
     body: Bytes,
+    user: Option<String>,
 ) -> Response {
     if !host.is_node_ready().await {
         return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
@@ -515,6 +517,7 @@ async fn start_kind<H: Host>(
         Supersede {
             force: force.clone(),
             active_peers: active_peers(&host).await,
+            by: user,
         },
         now,
         grace,
@@ -631,16 +634,18 @@ pub(crate) async fn get<H: Host>(
 /// `DELETE /admin/remakes/{id}`.
 pub(crate) async fn abort<H: Host>(
     State(host): State<H>,
+    axum::Extension(user): axum::Extension<crate::api::auth::AuthenticatedUser>,
     Path(id): Path<String>,
     Query(query): Query<AbortQuery>,
 ) -> Response {
-    abort_by_id(host, id, query).await
+    abort_by_id(host, id, query, Some(user.0)).await
 }
 
 /// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still
 /// undecided (proposed or expired), the oldest if there are several.
 pub(crate) async fn abort_current<H: Host>(
     State(host): State<H>,
+    axum::Extension(user): axum::Extension<crate::api::auth::AuthenticatedUser>,
     Query(query): Query<AbortQuery>,
 ) -> Response {
     let local = EpochRecord::from(host.current_epoch());
@@ -657,7 +662,7 @@ pub(crate) async fn abort_current<H: Host>(
         .await;
 
     match open {
-        Some(id) => abort_by_id(host, id, query).await,
+        Some(id) => abort_by_id(host, id, query, Some(user.0)).await,
         None => error_response(
             StatusCode::NOT_FOUND,
             "no remake of the local epoch is in progress",
@@ -666,7 +671,12 @@ pub(crate) async fn abort_current<H: Host>(
     }
 }
 
-async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Response {
+async fn abort_by_id<H: Host>(
+    host: H,
+    id: String,
+    query: AbortQuery,
+    user: Option<String>,
+) -> Response {
     let force = match Force::parse(query.force.as_deref(), ABORT_FORCE) {
         Ok(force) => force,
         Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None),
@@ -678,7 +688,7 @@ async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Respons
 
     match current.phase {
         Phase::Proposed => {}
-        Phase::Expired => return abort_expired(&host, current, &force).await,
+        Phase::Expired => return abort_expired(&host, current, &force, user).await,
         Phase::Committed => {
             return error_response(
                 StatusCode::CONFLICT,
@@ -750,7 +760,12 @@ async fn abort_by_id<H: Host>(host: H, id: String, query: AbortQuery) -> Respons
 /// Aborts expired remake `current` from this node: its coordinator is gone, and can't
 /// commit it any more (`coordinator`), unless `abort_obstacle` says otherwise. Answers
 /// with the final report.
-async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Response {
+async fn abort_expired<H: Host>(
+    host: &H,
+    current: Report,
+    force: &Force,
+    user: Option<String>,
+) -> Response {
     // Whether it expired is judged by this node's clock: one far ahead would abort a
     // remake its coordinator may still commit (a contested remake).
     if let Some(refused) = clock_obstacle(host, force, "whether the remake expired").await {
@@ -785,8 +800,10 @@ async fn abort_expired<H: Host>(host: &H, current: Report, force: &Force) -> Res
         decided_at: host.now_millis(),
     };
     let clock = host.clock();
+    // As the user who asks, who may write remakes (`api::access`): the rules take a
+    // decision but the coordinator's from one who may write it.
     let written = host
-        .commit_doc(move |doc| -> Result<bool, StoreError> {
+        .commit_doc_for(user, move |doc| -> Result<bool, StoreError> {
             let now = clock.now_millis();
             let expired = records::load(doc, &record_id).is_some_and(|record| {
                 record.phase(now, grace) == Phase::Expired && now >= abortable_at(&record, grace)
@@ -906,7 +923,8 @@ pub(crate) mod testing {
             CONTENT_TYPE,
             axum::http::HeaderValue::from_static("application/json"),
         );
-        into_json(start_kind(host, query, &headers, Bytes::from(body.to_string())).await).await
+        into_json(start_kind(host, query, &headers, Bytes::from(body.to_string()), None).await)
+            .await
     }
 
     /// `DELETE /admin/remakes/{id}` with `?force=`.
@@ -919,7 +937,21 @@ pub(crate) mod testing {
             wait: false,
             force: force.map(str::to_owned),
         };
-        into_json(abort_by_id(host, id.to_owned(), query).await).await
+        into_json(abort_by_id(host, id.to_owned(), query, None).await).await
+    }
+
+    /// `DELETE /admin/remakes/{id}` with `?force=`, asked by `user`.
+    pub(crate) async fn abort_as<H: Host>(
+        host: H,
+        id: &str,
+        force: Option<&str>,
+        user: &str,
+    ) -> (StatusCode, JsonValue) {
+        let query = AbortQuery {
+            wait: false,
+            force: force.map(str::to_owned),
+        };
+        into_json(abort_by_id(host, id.to_owned(), query, Some(user.to_owned())).await).await
     }
 
     /// `GET /admin/remakes/{id}`: the report, if the remake exists here.
diff --git a/src/remake/proposal.rs b/src/remake/proposal.rs
index 7695b4c..f8b3b0e 100644
--- a/src/remake/proposal.rs
+++ b/src/remake/proposal.rs
@@ -86,6 +86,9 @@ pub(super) struct ExpiredRemake {
 pub(super) struct Supersede {
     pub(super) force: Force,
     pub(super) active_peers: HashSet<String>,
+    /// Who asks (the API's user): the proposal, and the decisions aborting the remakes
+    /// it supersedes, are written as theirs, as the rules judge them (`access::rules`).
+    pub(super) by: Option<String>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -126,8 +129,9 @@ pub(super) async fn propose<H: Host>(
     let me = host.node_id();
     let started = host.remake().started_at().plus_ms(grace);
     let base_at_commit = kind::by_name(&proposal.kind).is_some_and(|kind| kind.base_at_commit());
+    let by = supersede.by.clone();
     let record = host
-        .commit_doc(move |doc| -> Result<Record, StartError> {
+        .commit_doc_for(by, move |doc| -> Result<Record, StartError> {
             let mut all = records::load_all(doc);
             if let Some((open, _)) = all.iter().find(|(_, record)| {
                 record.proposal.from_epoch == proposal.from_epoch
diff --git a/src/simtest/policy.rs b/src/simtest/policy.rs
index 9f38e44..4162a8c 100644
--- a/src/simtest/policy.rs
+++ b/src/simtest/policy.rs
@@ -398,7 +398,8 @@ async fn a_strict_swarm() {
 }
 
 /// 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.
+/// expired a grace period ago, any member aborts it, asked by a user who may write
+/// remakes, 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;
@@ -456,7 +457,8 @@ async fn an_expired_remake_aborted_by_any_member() {
         "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;
+            let (status, _) =
+                crate::remake::testing::abort_as(b_ref.clone(), id_ref, None, "alice").await;
             status == StatusCode::OK
         },
     )