@@ -63,7 +63,7 @@ jobs:
       # lag leanVM, and `--all-targets` leaves such targets out. Name them so
       # they keep compiling until they run again.
       - name: Clippy (skipped spectests)
-        run: cargo clippy --locked --workspace --test forkchoice_spectests --test signature_spectests --test stf_spectests --test ssz_spectests -- -D warnings
+        run: cargo clippy --locked --workspace --test forkchoice_spectests --test signature_spectests --test stf_spectests --test ssz_spectests --test interchange_vectors -- -D warnings
 
   # Two features gate code that no step in `lint` reaches, so both would rot
   # silently between merges.
 
@@ -55,9 +55,12 @@ pub(crate) fn run(options: SlashingProtectionOptions) -> eyre::Result<()> {
         } => {
             let interchange = std::fs::read_to_string(&file)
                 .map_err(|err| eyre::eyre!("{}: {err}", file.display()))?;
-            let chain = SlashingProtection::interchange_chain(&interchange)?;
-            let mut protection = SlashingProtection::open_offline(&validators_dir, Some(chain))?;
-            let validators = protection.import(&interchange)?;
+            // Parsed once: the file can be tens of megabytes in the complete
+            // format, and its chain is needed before the record is opened.
+            let staged = SlashingProtection::stage(&interchange)?;
+            let mut protection =
+                SlashingProtection::open_offline(&validators_dir, Some(staged.root()))?;
+            let (validators, _) = protection.absorb(Some(&staged), &[])?;
             info!(
                 validators,
                 file = %file.display(),
 
@@ -151,7 +151,7 @@ pub async fn list(State(context): State<KeymanagerContext>) -> Json<ListResponse
 
 pub async fn import(
     State(context): State<KeymanagerContext>,
-    Json(request): Json<ImportRequest>,
+    Json(mut request): Json<ImportRequest>,
 ) -> Json<StatusResponse> {
     if request.keystores.len() > MAX_KEYSTORES_PER_IMPORT {
         let message = format!(
@@ -174,11 +174,23 @@ pub async fn import(
     // record's chain before anything is touched, so a request whose history
     // would be refused imports no key at all, rather than some keys without
     // it. It is written later, in the same write as the keys' registration,
-    // once the keys it came with are known to be saved.
-    let history = match &request.slashing_protection {
+    // once the keys it came with are known to be saved. Parsed on the
+    // blocking pool: a complete-format interchange runs to tens of megabytes
+    // (see `KEYMANAGER_BODY_LIMIT`), and the executor thread this handler
+    // runs on also drives the duty services.
+    let history = match request.slashing_protection.take() {
         None => None,
         Some(interchange) => {
-            let staged = crate::slashing_protection::SlashingProtection::stage(interchange);
+            let staged = tokio::task::spawn_blocking(move || {
+                crate::slashing_protection::SlashingProtection::stage(&interchange)
+            })
+            .await
+            .unwrap_or_else(|err| {
+                Err(Error::SlashingProtection {
+                    path: "interchange".to_string(),
+                    reason: format!("could not be parsed: {err}"),
+                })
+            });
             let judged = match staged {
                 Ok(staged) => {
                     let staged = Arc::new(staged);
@@ -573,10 +585,14 @@ pub async fn delete(
 
     // Exported only now, with the keys out of the store: a signature with one
     // of them either finished before the write lock above was granted, and is
-    // in the record, or can no longer happen. Only for the keys that are
-    // `deleted` or may become `not_active`, as the specification requires: a
-    // key whose delete failed is still on disk and comes back on restart, so
-    // its history is not handed over. Teku and Prysm filter the same way.
+    // in the record, or can no longer happen. That holds because the duty
+    // services record before they take the store's read lock to sign (see
+    // `attestation.rs` and `proposal.rs`), so no signature exists without its
+    // record already on disk. Only for the keys that are `deleted` or may
+    // become `not_active`, as the specification requires: a key whose delete
+    // failed is still on disk and comes back on restart, so its history is
+    // not handed over. Teku and Prysm filter the same way. A key listed twice
+    // is `deleted` once and `not_found` once; `export` lists it once.
     let requested: Vec<Option<BlsPubkey>> = results
         .iter()
         .map(|(pubkey, status)| match status.status.as_str() {
@@ -601,7 +617,11 @@ pub async fn delete(
             // history from one whose history could not be read, so every key
             // that parsed becomes an error, as Prysm answers: a `not_found`
             // here would tell a migrator there is nothing to carry. A
-            // malformed pubkey keeps its own error.
+            // malformed pubkey keeps its own error. The deletions themselves
+            // already happened and are not undone: the keys' records are
+            // kept, so a retry answers `not_active` with their history. The
+            // field is required, so it carries an empty string rather than
+            // an interchange that would read as "nothing signed".
             warn!(%err, "Deleted keys whose slashing-protection history could not be exported");
             let message = format!("slashing_protection could not be exported: {err}");
             let data = results
@@ -1280,10 +1300,14 @@ mod tests {
 
     /// A minimal interchange for `KEYSTORE_PUBKEY` on the chain `root`.
     fn history(root: &str, slot: u64, source: u64, target: u64) -> String {
+        history_for(KEYSTORE_PUBKEY, root, slot, source, target)
+    }
+
+    fn history_for(pubkey: &str, root: &str, slot: u64, source: u64, target: u64) -> String {
         serde_json::json!({
             "metadata": { "interchange_format_version": "5", "genesis_validators_root": root },
             "data": [{
-                "pubkey": KEYSTORE_PUBKEY,
+                "pubkey": pubkey,
                 "signed_blocks": [{ "slot": slot.to_string() }],
                 "signed_attestations": [{
                     "source_epoch": source.to_string(),
@@ -1425,6 +1449,94 @@ mod tests {
         );
     }
 
+    /// The history may name a validator that is not in the request: it gets a
+    /// record too, so a later delete of it answers `not_active` with that
+    /// history rather than `not_found`.
+    #[tokio::test]
+    async fn history_for_a_key_outside_the_request_gets_a_record() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let (context, protection) = context_with_record(dir.path());
+        let app = router(context, TOKEN.to_string());
+        let other = format!("0x{}", "ab".repeat(48));
+
+        let (_, body) = send(
+            app.clone(),
+            authed(
+                "POST",
+                "/eth/v1/keystores",
+                serde_json::json!({
+                    "keystores": [keystore_json()],
+                    "passwords": [PASSWORD],
+                    "slashing_protection": history_for(&other, ZERO_ROOT, 100, 5, 10),
+                }),
+            ),
+        )
+        .await;
+        assert_eq!(body["data"][0]["status"], "imported", "got {body}");
+
+        let other_pubkey = parse_pubkey(&other).expect("pubkey");
+        {
+            let protection = protection.lock().expect("not poisoned");
+            assert!(protection.contains(&other_pubkey));
+            assert!(protection.check_block(&other_pubkey, 100).is_err());
+            assert!(protection.contains(&parse_pubkey(KEYSTORE_PUBKEY).expect("pubkey")));
+        }
+
+        let (_, body) = send(
+            app,
+            authed(
+                "DELETE",
+                "/eth/v1/keystores",
+                serde_json::json!({ "pubkeys": [other] }),
+            ),
+        )
+        .await;
+        assert_eq!(body["data"][0]["status"], "not_active", "got {body}");
+        let interchange: serde_json::Value =
+            serde_json::from_str(body["slashing_protection"].as_str().expect("a string"))
+                .expect("json");
+        assert_eq!(interchange["data"][0]["pubkey"], other);
+        assert_eq!(interchange["data"][0]["signed_blocks"][0]["slot"], "100");
+    }
+
+    /// A key listed twice in one delete is `deleted` once and `not_active`
+    /// once, and the returned interchange names it once: one that named a
+    /// validator twice would be refused on import.
+    #[tokio::test]
+    async fn a_delete_listing_a_key_twice_exports_it_once() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let (context, _) = context_with_record(dir.path());
+        let app = router(context, TOKEN.to_string());
+
+        let (_, body) = send(
+            app.clone(),
+            authed(
+                "POST",
+                "/eth/v1/keystores",
+                serde_json::json!({ "keystores": [keystore_json()], "passwords": [PASSWORD] }),
+            ),
+        )
+        .await;
+        assert_eq!(body["data"][0]["status"], "imported", "got {body}");
+
+        let (_, body) = send(
+            app,
+            authed(
+                "DELETE",
+                "/eth/v1/keystores",
+                serde_json::json!({ "pubkeys": [KEYSTORE_PUBKEY, KEYSTORE_PUBKEY] }),
+            ),
+        )
+        .await;
+        assert_eq!(body["data"][0]["status"], "deleted", "got {body}");
+        assert_eq!(body["data"][1]["status"], "not_active", "got {body}");
+        let interchange: serde_json::Value =
+            serde_json::from_str(body["slashing_protection"].as_str().expect("a string"))
+                .expect("json");
+        assert_eq!(interchange["data"].as_array().expect("array").len(), 1);
+        assert_eq!(interchange["data"][0]["pubkey"], KEYSTORE_PUBKEY);
+    }
+
     /// History that came with a key that then failed to import is not
     /// written: nothing it describes is held here.
     #[cfg(unix)]
 
@@ -59,7 +59,7 @@
 
 pub mod file;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 
@@ -150,6 +150,14 @@ pub struct StagedHistory {
     entries: Vec<(BlsPubkey, Record)>,
 }
 
+impl StagedHistory {
+    /// The chain the interchange names: what a record created for it must
+    /// belong to.
+    pub fn root(&self) -> Root {
+        self.root
+    }
+}
+
 /// What one validator has signed: its watermarks.
 ///
 /// `None` is "never signed", kept distinct from zero so that slot 0 and epoch
@@ -531,17 +539,6 @@ impl SlashingProtection {
         self.absorb(None, pubkeys).map(|(_, registered)| registered)
     }
 
-    /// The chain an EIP-3076 interchange names, read without importing it: what
-    /// a record created for it must belong to.
-    pub fn interchange_chain(interchange: &str) -> Result<Root> {
-        file::parse(interchange)
-            .map(|(root, _)| root)
-            .map_err(|reason| Error::SlashingProtection {
-                path: "interchange".to_string(),
-                reason,
-            })
-    }
-
     /// The chain this record belongs to.
     pub fn genesis_validators_root(&self) -> Root {
         self.genesis_validators_root
@@ -607,6 +604,11 @@ impl SlashingProtection {
     /// every key in `register` that then still has no record, in one durable
     /// write: either all of it is on disk or none of it is. Returns how many
     /// validators the history named and how many keys were newly registered.
+    ///
+    /// A key in `register` that already has a record and is not in the
+    /// history keeps that record, with a warning: it is what this client last
+    /// signed with the key, which says nothing about what the key signed
+    /// elsewhere since.
     pub fn absorb(
         &mut self,
         history: Option<&StagedHistory>,
@@ -627,10 +629,19 @@ impl SlashingProtection {
         let named = staged.len();
         let mut registered = 0;
         for pubkey in register {
-            if !staged.contains_key(pubkey) && !self.records.contains_key(pubkey) {
-                staged.insert(*pubkey, Record::default());
-                registered += 1;
+            if staged.contains_key(pubkey) {
+                continue;
             }
+            if self.records.contains_key(pubkey) {
+                warn!(
+                    pubkey = %encode_hex(&pubkey.0),
+                    "Registering a validator that already has a slashing-protection record and \
+                     brought no history; it continues from what this client last signed with it"
+                );
+                continue;
+            }
+            staged.insert(*pubkey, Record::default());
+            registered += 1;
         }
         if !staged.is_empty() {
             self.commit(staged)?;
@@ -642,10 +653,14 @@ impl SlashingProtection {
     ///
     /// A key with no record is left out rather than listed as never having
     /// signed: an empty entry would tell the importer something this client
-    /// does not know.
+    /// does not know. A key listed twice is exported once: an interchange that
+    /// names a validator twice is refused by stricter importers, this client's
+    /// own record reader among them.
     pub fn export(&self, pubkeys: &[BlsPubkey]) -> Result<String> {
+        let mut seen = HashSet::with_capacity(pubkeys.len());
         let selected = pubkeys
             .iter()
+            .filter(|pubkey| seen.insert(**pubkey))
             .filter_map(|pubkey| self.records.get_key_value(pubkey));
         file::render(self.genesis_validators_root, selected).map_err(|err| {
             Error::SlashingProtection {
@@ -1189,6 +1204,30 @@ mod tests {
         assert!(attest(&mut protection, 1, &data(128, 3, 4, 0xaa)).is_ok());
     }
 
+    /// A newer history imported over an older record raises it, and the raised
+    /// watermarks survive a restart: the merge updates an existing entry, not
+    /// only inserts a missing one.
+    #[test]
+    fn an_import_raises_an_existing_watermark() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        {
+            let mut protection = open(dir.path(), &[1], true).expect("init");
+            propose(&mut protection, 1, 10).expect("propose");
+            attest(&mut protection, 1, &data(96, 2, 3, 0xaa)).expect("attest");
+
+            protection
+                .import(&interchange(0x42, 1, 50, 4, 6))
+                .expect("imports");
+            assert!(propose(&mut protection, 1, 40).is_err());
+            assert!(attest(&mut protection, 1, &data(160, 3, 5, 0xaa)).is_err());
+        }
+        let mut protection = open(dir.path(), &[1], false).expect("reopens");
+        assert!(propose(&mut protection, 1, 50).is_err());
+        assert!(propose(&mut protection, 1, 51).is_ok());
+        assert!(attest(&mut protection, 1, &data(192, 4, 6, 0xaa)).is_err());
+        assert!(attest(&mut protection, 1, &data(224, 4, 7, 0xaa)).is_ok());
+    }
+
     /// An older history imported over a newer record must not lower it.
     #[test]
     fn an_import_never_lowers_a_watermark() {
@@ -1233,6 +1272,24 @@ mod tests {
         assert!(propose(&mut fresh, 1, 97).is_err());
     }
 
+    /// A key requested twice is exported once: this client's own record
+    /// reader refuses a file naming a validator twice, and so may others'.
+    #[test]
+    fn an_export_lists_a_key_requested_twice_once() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let mut protection = open(dir.path(), &[1], true).expect("init");
+        propose(&mut protection, 1, 97).expect("propose");
+
+        let exported = protection.export(&[pubkey(1), pubkey(1)]).expect("exports");
+        let json: serde_json::Value = serde_json::from_str(&exported).expect("json");
+        assert_eq!(json["data"].as_array().expect("array").len(), 1);
+
+        let other = tempfile::tempdir().expect("temp dir");
+        let mut fresh = open(other.path(), &[], false).expect("opens");
+        fresh.import(&exported).expect("round trip");
+        assert!(propose(&mut fresh, 1, 97).is_err());
+    }
+
     #[test]
     fn opening_offline_needs_a_record_or_a_chain_to_create_one_for() {
         let dir = tempfile::tempdir().expect("temp dir");
@@ -1276,7 +1333,7 @@ mod tests {
         let mut protection = open(dir.path(), &[], false).expect("opens");
         for text in ["", "{}", "[]"] {
             protection.import(text).expect_err("not an interchange");
-            SlashingProtection::interchange_chain(text).expect_err("not an interchange");
+            SlashingProtection::stage(text).expect_err("not an interchange");
         }
         assert!(protection.is_empty());
     }
 
@@ -15,7 +15,7 @@ use std::collections::HashSet;
 use std::path::{Path, PathBuf};
 
 use ethlambda_types::beacon::containers::shared::{AttestationData, Checkpoint};
-use ethlambda_types::beacon::primitives::{BlsPubkey, H256, Root};
+use ethlambda_types::beacon::primitives::{BlsPubkey, Root};
 use ethlambda_validator::beacon_node::dto::parse_pubkey;
 use ethlambda_validator::slashing_protection::SlashingProtection;
 use serde::Deserialize;
 
@@ -470,7 +470,7 @@ ethlambda slashing-protection export --validators-dir ./validators --pubkeys 0xa
 
 | Sub-command | Does |
 |---|---|
-| `import <file>` | Merges a version 5 interchange, complete or minimal, into the record. Each validator is reduced to its highest slot, source and target and merged by maximum, so nothing already recorded is lowered and slashable data in the file is absorbed rather than refused. All or nothing. With no record yet, the import creates one for the interchange's chain, and a failed import creates nothing but the lock file; the client refuses the record at its next start if that is not its beacon node's chain |
+| `import <file>` | Merges a version 5 interchange, complete or minimal, into the record. Each validator is reduced to its highest slot, source and target and merged by maximum, so nothing already recorded is lowered and slashable data in the file is absorbed rather than refused. All or nothing. With no record yet, the import creates one for the interchange's chain, and a failed import creates nothing; the client refuses the record at its next start if that is not its beacon node's chain |
 | `export [--pubkeys ...] <file>` | Writes the record, or the named validators' part of it, as a minimal version 5 interchange, readable by its owner only. A named validator with no record fails the export, so a typo or the wrong directory cannot produce a file that reads as complete. Takes the record's lock, so the directory must be writable. Refuses to overwrite an existing file |
 
 ## `benchmark import` flags