remake: replay from the quarantine where the user may write

POST /admin/quarantine/{id}/replay needs write on _continuum, which
never covers user data, and replayed every path of the entry as the
user who asked: every node refused the user data in it. It now
replays only the paths the user may write and keeps the rest in the
entry (left), for one who may; 403 if the user may write none.

Review finding 3 (quarantine replays).

3 files changed, 51 insertions, 16 deletions

diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index d3715b1..960ab14 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1705,8 +1705,9 @@ paths:
       summary: Replay quarantined writes
       description: |
         Writes them into the current epoch at their old paths, as fresh writes (they
-        win over what is there), and removes the entry. If none can be written, the
-        entry stays.
+        win over what is there), and removes the entry. Only paths the user may write
+        (every node judges the change by them): the rest stay in the entry (`left`), for
+        one who may. If none can be written, the entry stays.
       responses:
         '200':
           description: Replayed.
@@ -1714,14 +1715,23 @@ paths:
             application/json:
               schema:
                 type: object
-                required: [replayed, failed]
+                required: [replayed, failed, left]
                 properties:
                   replayed:
                     type: integer
                   failed:
                     type: integer
+                  left:
+                    type: integer
+                    description: Paths the user may not write, left in the entry.
         '401':
           $ref: '#/components/responses/Unauthorized'
+        '403':
+          description: The user may write none of its paths.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
         '404':
           $ref: '#/components/responses/Error'
         '409':
diff --git a/docs/remakes.md b/docs/remakes.md
index cad429f..8e585c6 100644
--- a/docs/remakes.md
+++ b/docs/remakes.md
@@ -339,7 +339,9 @@ curl -X DELETE localhost:3000/admin/quarantine/<id>       # discard
 ```
 
 Replaying writes the values at their old paths as fresh writes (they win over what is
-there now), in the name of the user who asked. Quarantine doesn't survive a restart of that node, so look at it soon; the
+there now), in the name of the user who asked: only at the paths it may write, as every
+node judges the change by them ([`access.md`](access.md)); the rest stay (`left`), for
+one who may. Quarantine doesn't survive a restart of that node, so look at it soon; the
 remake report's `local.quarantine` on that node names the entry.
 
 ## Troubleshooting
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
index e7214d6..24262fc 100644
--- a/src/remake/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -177,21 +177,29 @@ pub(crate) async fn replay<H: Host>(
     let replayed = {
         let entry = Arc::clone(&entry);
         let user = user.map(|axum::Extension(user)| user.0);
-        host.commit_doc_for(user, move |doc| {
+        host.commit_doc_for(user.clone(), move |doc| {
+            // Only where the operator may write, as every node judges the change by it
+            // (`access::rules`); the rest stays here.
+            let caps = user
+                .as_deref()
+                .map(|user| crate::access::registry::caps_of(&*doc, user));
+            let registry = crate::access::registry::exists(&*doc);
+            let (paths, kept): (Vec<_>, Vec<_>) = entry.paths.iter().cloned().partition(|path| {
+                !registry
+                    || caps
+                        .as_ref()
+                        .is_some_and(|caps| caps.allows(crate::access::Right::Write, path))
+            });
             // As they are: the operator decided they belong at their old paths.
-            Ok::<_, StoreError>(delta::replay_paths(
-                &entry.document,
-                doc,
-                &entry.paths,
-                &Translator::default(),
-            ))
+            let stats = delta::replay_paths(&entry.document, doc, &paths, &Translator::default());
+            Ok::<_, StoreError>((stats, kept))
         })
         .await
     };
     // The commit's task held the only other reference, and it is done.
-    let entry = Arc::into_inner(entry).expect("the replay's reference is gone");
-    let stats = match replayed {
-        Ok(stats) => stats,
+    let mut entry = Arc::into_inner(entry).expect("the replay's reference is gone");
+    let (stats, kept) = match replayed {
+        Ok(replayed) => replayed,
         Err(error) => {
             // Rejoining since the check above: keep the entry for later.
             host.remake().restore_quarantined(entry);
@@ -199,6 +207,14 @@ pub(crate) async fn replay<H: Host>(
         }
     };
 
+    if stats.replayed == 0 && !kept.is_empty() && kept.len() == entry.paths.len() {
+        host.remake().restore_quarantined(entry);
+        return error_response(
+            StatusCode::FORBIDDEN,
+            "the user may write none of its paths",
+            None,
+        );
+    }
     if stats.failed > 0 && stats.replayed == 0 {
         // Nothing went in; keep the entry for another look.
         host.remake().restore_quarantined(entry);
@@ -208,9 +224,16 @@ pub(crate) async fn replay<H: Host>(
             Some(json!({ "failed": stats.failed })),
         );
     }
+    let left = kept.len();
+    if !kept.is_empty() {
+        // Those the user may not write, for one who may.
+        entry.paths = kept;
+        host.remake().restore_quarantined(entry);
+    }
 
-    info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, "Replayed quarantined changes");
-    Json(json!({ "replayed": stats.replayed, "failed": stats.failed })).into_response()
+    info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, left, "Replayed quarantined changes");
+    Json(json!({ "replayed": stats.replayed, "failed": stats.failed, "left": left }))
+        .into_response()
 }
 
 pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {