From 3db2e841e21781a283aeedd925f08e420685d39f Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Wed, 11 Nov 2026 17:54:06 -0300 Subject: [PATCH 1/6] Address the last review round on the slashing-protection PRs (#734, #736) - A re-import whose registration failed and was fully undone now answers "not imported: ...; its keystore and definitions entry are back as they were". "keystore rewritten" stopped being true once the undo put the files back too, and is kept for the case where the undo itself failed. - The interchange vector runner ends a case only when the refused step holds an attestation whose source is at or past its target outside genesis, the one shape this importer refuses, rather than on any step flagged as slashable. Every other slashable step must now import. - an_import_carrying_history_is_refused_whole is renamed for what it has tested since the importer landed, history that does not parse, and asserts that this is the reason. - The Interchange types' doc says they are shared by the record reader and the importer, both strict. - docs/cli.md no longer says an export needs a writable directory: the lock is on the directory itself and an export writes nothing there. A test pins it on a mode 0500 directory. --- crates/validator/src/http_api/keystores.rs | 64 +++++++++++-------- .../validator/src/slashing_protection/file.rs | 11 ++-- .../validator/src/slashing_protection/mod.rs | 25 ++++++++ crates/validator/tests/interchange_vectors.rs | 34 +++++++++- docs/cli.md | 2 +- 5 files changed, 102 insertions(+), 34 deletions(-) diff --git a/crates/validator/src/http_api/keystores.rs b/crates/validator/src/http_api/keystores.rs index c8356decb..72dbbe77e 100644 --- a/crates/validator/src/http_api/keystores.rs +++ b/crates/validator/src/http_api/keystores.rs @@ -407,31 +407,31 @@ pub async fn import( .map(|result| match (result, ®istration) { (Ok((_, secret, _)), Ok(_)) => finish_import(&mut store, &secret), (Ok((pubkey, _, _)), Err(err)) => { - if before.contains_key(&pubkey) { - // Its keystore was rewritten; its entry is back as it was - // and nothing about it changed in memory, so it stays - // whatever it was, loaded since startup in the ordinary - // case. - let entry = if rolled_back.is_ok() { - "its definitions entry is back as it was" - } else { - "its definitions entry could not be put back" - }; - KeyStatus::error(format!( + match (before.contains_key(&pubkey), rolled_back.is_ok()) { + // A key the file already listed, undone: its keystore, + // its password and its entry are what they were before + // the request, and nothing about it changed in memory, so + // it stays whatever it was, loaded since startup in the + // ordinary case. + (true, true) => KeyStatus::error(format!( + "not imported: its slashing-protection record could not be written: \ + {err}; its keystore and definitions entry are back as they were" + )), + // The undo failed as well, so what this request wrote + // over the key's files and entry may still be there. + (true, false) => KeyStatus::error(format!( "keystore rewritten, but the slashing-protection record could not be \ - written: {err}; {entry}" - )) - } else if rolled_back.is_err() { + written: {err}; its definitions entry could not be put back" + )), + (false, true) => KeyStatus::error(format!( + "not imported: its slashing-protection record could not be written: {err}" + )), // Still on disk, not loaded now, and refused at the next // start until it has a record. - KeyStatus::error(format!( + (false, false) => KeyStatus::error(format!( "saved, but not loaded: its slashing-protection record could not be \ written: {err}" - )) - } else { - KeyStatus::error(format!( - "not imported: its slashing-protection record could not be written: {err}" - )) + )), } } (Err(status), _) => status, @@ -1508,11 +1508,12 @@ mod tests { ) .await; assert_eq!(body["data"][0]["status"], "error", "got {body}"); + // Not "keystore rewritten": the undo put the files back too, so the + // message must not send the operator looking for a new keystore. + let message = body["data"][0]["message"].as_str().unwrap_or_default(); + assert!(message.starts_with("not imported"), "got {body}"); assert!( - body["data"][0]["message"] - .as_str() - .unwrap_or_default() - .contains("back as it was"), + message.ends_with("its keystore and definitions entry are back as they were"), "got {body}" ); let definitions = ValidatorDefinitions::open(dir.path()).expect("opens"); @@ -1637,9 +1638,12 @@ mod tests { assert_eq!(definitions.0, vec![previous]); } - /// A request that carries history is refused whole, importing nothing. + /// A request whose history does not parse as an interchange is refused + /// whole, importing nothing: `{}` has neither metadata nor data. History + /// that does parse is imported with its keys (see + /// `an_import_brings_its_slashing_protection_history`). #[tokio::test] - async fn an_import_carrying_history_is_refused_whole() { + async fn an_import_whose_history_does_not_parse_is_refused_whole() { let dir = tempfile::tempdir().expect("temp dir"); let context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); let store = context.store.clone(); @@ -1658,6 +1662,14 @@ mod tests { ) .await; assert_eq!(body["data"][0]["status"], "error", "got {body}"); + // Refused for the history, and for its shape: not for the keystore, + // and not for some later step. + let message = body["data"][0]["message"].as_str().unwrap_or_default(); + assert!( + message.starts_with("slashing_protection was not imported") + && message.contains("malformed"), + "got {body}" + ); assert!(store.read().await.pubkeys().is_empty()); assert!( ValidatorDefinitions::open(dir.path()) diff --git a/crates/validator/src/slashing_protection/file.rs b/crates/validator/src/slashing_protection/file.rs index 865a8237c..9018e4aba 100644 --- a/crates/validator/src/slashing_protection/file.rs +++ b/crates/validator/src/slashing_protection/file.rs @@ -42,10 +42,13 @@ pub const FILE_NAME: &str = "slashing_protection.json"; /// The only EIP-3076 interchange version this client reads or writes. pub const INTERCHANGE_VERSION: &str = "5"; -/// This client's own file, read strictly: a field it did not write is an -/// error, since nothing else should have written here. Not for reading -/// other clients' interchange files, which may carry fields this refuses; -/// an importer needs its own, lenient, types. +/// An EIP-3076 version 5 interchange: the shape of this client's own record +/// and of every file it imports. The record reader ([`read`]) and the +/// importer ([`parse`], which `SlashingProtection::stage` calls) share these +/// types, and both read strictly: a field the format does not define is an +/// error rather than ignored, as it is in Lighthouse's EIP-3076 types, so a +/// file that is not the interchange it claims to be is refused whole instead +/// of half read. #[derive(Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct Interchange { diff --git a/crates/validator/src/slashing_protection/mod.rs b/crates/validator/src/slashing_protection/mod.rs index 04f7c861d..909d14e32 100644 --- a/crates/validator/src/slashing_protection/mod.rs +++ b/crates/validator/src/slashing_protection/mod.rs @@ -1459,6 +1459,31 @@ mod tests { assert!(propose(&mut fresh, 1, 97).is_err()); } + /// An export writes nothing to the validators directory, and the lock is + /// an `flock` on the directory itself, which a read-only descriptor can + /// take: keys mounted read-only can still hand their history over. + #[cfg(unix)] + #[test] + fn an_export_needs_no_writable_directory() { + use std::os::unix::fs::PermissionsExt as _; + + let dir = tempfile::tempdir().expect("temp dir"); + let mut protection = open(dir.path(), &[1], true).expect("init"); + propose(&mut protection, 1, 97).expect("propose"); + drop(protection); + + let read_only = std::fs::Permissions::from_mode(0o500); + std::fs::set_permissions(dir.path(), read_only).expect("chmod"); + let exported = SlashingProtection::open_offline(dir.path(), None) + .and_then(|protection| protection.export(&[pubkey(1)])); + let writable = std::fs::Permissions::from_mode(0o700); + std::fs::set_permissions(dir.path(), writable).expect("chmod back"); + + let exported = exported.expect("locks, reads and exports with no write access"); + let json: serde_json::Value = serde_json::from_str(&exported).expect("json"); + assert_eq!(json["data"][0]["signed_blocks"][0]["slot"], "97"); + } + #[test] fn opening_offline_needs_a_record_or_a_chain_to_create_one_for() { let dir = tempfile::tempdir().expect("temp dir"); diff --git a/crates/validator/tests/interchange_vectors.rs b/crates/validator/tests/interchange_vectors.rs index a338d8b73..4823dd51d 100644 --- a/crates/validator/tests/interchange_vectors.rs +++ b/crates/validator/tests/interchange_vectors.rs @@ -128,9 +128,17 @@ fn run(case: &Case) -> Result<(), String> { // The README lets an importer refuse an interchange holding // slashable data, in which case the rest of the case is skipped. // This one absorbs all of it but an attestation whose source is at - // or past its target, so a refusal on such a step ends the case, - // as it does in Lighthouse; on any other step it is a failure. - (false, true) if step.contains_slashable_data => return Ok(()), + // or past its target, so only a step whose interchange holds one + // may be refused, which ends the case, as it does in Lighthouse. + // Every other slashable step must import: skipping on the flag + // alone would pass a suite whose importer had started refusing, + // say, a duplicated pubkey or a surround, with that case's blocks + // and attestations never attempted. + (false, true) + if step.contains_slashable_data && holds_impossible_source(&step.interchange) => + { + return Ok(()); + } (false, true) => { return Err(format!( "step {index}: import failed but should succeed (slashable data: {}): {}", @@ -182,6 +190,26 @@ fn run(case: &Case) -> Result<(), String> { Ok(()) } +/// Whether `interchange` holds an attestation whose source is at or past its +/// target outside genesis: the one shape this importer refuses rather than +/// absorbs. The rule is restated here rather than borrowed from the crate, so +/// that the importer widening what it refuses cannot widen what this runner +/// forgives along with it. +fn holds_impossible_source(interchange: &serde_json::Value) -> bool { + let epoch = + |value: &serde_json::Value| value.as_str().and_then(|text| text.parse::().ok()); + interchange["data"] + .as_array() + .into_iter() + .flatten() + .flat_map(|entry| entry["signed_attestations"].as_array().into_iter().flatten()) + .any(|attestation| { + let source = epoch(&attestation["source_epoch"]); + let target = epoch(&attestation["target_epoch"]); + matches!((source, target), (Some(source), Some(target)) if source >= target && source != 0) + }) +} + /// Only the epochs are judged; the rest of the data is filler. fn data(source: u64, target: u64) -> AttestationData { AttestationData { diff --git a/docs/cli.md b/docs/cli.md index e77aa426a..5f75850f5 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -474,7 +474,7 @@ ethlambda slashing-protection export --validators-dir ./validators --pubkeys 0xa | Sub-command | Does | |---|---| | `import ` | 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, except an attestation whose source is at or past its target outside genesis, which is 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 ...] ` | 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 | +| `export [--pubkeys ...] ` | 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. Writes nothing to the directory, and the lock is on the directory itself, so a read-only one will do. Refuses to overwrite an existing file | ## `benchmark import` flags From 573395d374413c522bee9b5870a05f11d35b4376 Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Wed, 11 Nov 2026 17:58:27 -0300 Subject: [PATCH 2/6] Refuse attestation data whose index cannot be valid for its fork validate_attestation_data checked the slot, the target epoch and the source, and passed data.index through unchecked. It now takes the slot's fork and refuses any index but 0 before gloas and any above 1 from gloas, where the index is the payload signal. Those are gossip's own bounds (attestation.data.index == 0 from electra, < 2 from gloas), and Lodestar's validator client refuses the same before signing. It is a liveness check rather than a slashing one. A vote whose index cannot be valid is rejected by every peer, after slashing protection has recorded its target, so the validator has no second vote to cast in that epoch. Refused in the beacon-node implementation, the answer is an Err that failover moves past, and nothing is recorded for it. Before electra the index is a committee, and this client only ever asks about committee 0. At gloas only the range is checked, since which of 0 and 1 is right depends on the node's view of the block. Every fork is a named match arm, so a fork after gloas has to say what its index means. --- crates/validator/src/attestation.rs | 40 +++++- crates/validator/src/beacon_node/fallback.rs | 23 ++++ crates/validator/src/beacon_node/http.rs | 2 +- crates/validator/src/beacon_node/mock.rs | 2 +- crates/validator/src/beacon_node/mod.rs | 138 +++++++++++++++++-- docs/cli.md | 8 +- 6 files changed, 194 insertions(+), 19 deletions(-) diff --git a/crates/validator/src/attestation.rs b/crates/validator/src/attestation.rs index 5372b8470..9de0b770b 100644 --- a/crates/validator/src/attestation.rs +++ b/crates/validator/src/attestation.rs @@ -135,7 +135,7 @@ impl AttestationService { // returned; failing here loses the slot rather than trying the next // node. That is why the same check now lives in the implementation, // where an `Err` is something failover can act on. - validate_attestation_data(slot, &data)?; + validate_attestation_data(slot, slot_fork, &data)?; // Consistent with the check just above: `data.target.epoch` is now // known to equal `expected_target_epoch`, so this is the fork this @@ -763,6 +763,44 @@ mod tests { assert!(node.submitted().is_empty()); } + /// A liveness check rather than a slashing one: an index that cannot be + /// valid at the slot's fork is refused before anything is recorded, so + /// the validator's one vote for the epoch is not spent on an attestation + /// every peer would reject, and a correct answer for the same slot is + /// still signed. `context()` is mainnet's schedule, where slot 96 is long + /// before gloas, so 1 is not an index a vote there can carry. + #[tokio::test] + async fn an_index_invalid_at_the_slots_fork_is_refused_and_records_nothing() { + let mut store = ValidatorStore::new(); + let pubkey = store.insert_secret("test", &secret()).expect("inserts"); + let duties = vec![duty(&encode_hex(&pubkey.0), 1337, 96, 3)]; + let store = RwLock::new(store); + let protection = SlashingProtection::in_memory(); + + let wrong = Arc::new( + MockBeaconNode::new() + .with_attestation_data(96) + .with_attestation_index(1), + ); + let err = AttestationService::new(wrong.clone(), context(), protection.clone()) + .attest(96, &duties, &store) + .await + .expect_err("must reject an index of 1 before gloas"); + assert!(matches!(err, Error::InconsistentResponse(_)), "got {err:?}"); + assert!(wrong.submitted().is_empty(), "nothing may be signed"); + + let right = Arc::new(MockBeaconNode::new().with_attestation_data(96)); + let attested = AttestationService::new(right.clone(), context(), protection) + .attest(96, &duties, &store) + .await + .expect("attests"); + assert_eq!( + (attested.published, attested.refused), + (1, 0), + "the refused answer must not have recorded the target" + ); + } + /// The regression test for why `attest` derives the `Eth-Consensus-Version` /// header from `data.target.epoch` rather than accepting it from the /// caller: the two slots below are one epoch apart and straddle mainnet's diff --git a/crates/validator/src/beacon_node/fallback.rs b/crates/validator/src/beacon_node/fallback.rs index f05107ab0..918386c50 100644 --- a/crates/validator/src/beacon_node/fallback.rs +++ b/crates/validator/src/beacon_node/fallback.rs @@ -501,6 +501,29 @@ mod tests { assert!(matches!(err, Error::AllBeaconNodesFailed(_)), "got {err:?}"); } + /// The same arrangement for the index: a node answering with one no vote + /// at the slot's fork can carry is failed over from, rather than its + /// answer being signed, recorded and then rejected by every peer. + #[tokio::test] + async fn a_node_answering_with_an_index_invalid_at_the_fork_is_failed_over_from() { + // Electra fixed the index at zero; node 1 answers 1 regardless. + let wrong = MockBeaconNode::new() + .with_attestation_data(96) + .with_attestation_index(1); + let right = MockBeaconNode::new().with_attestation_data(96); + let fallback = FallbackBeaconNode::new(vec![wrong, right]); + + let data = fallback + .attestation_data(96, ForkName::Electra) + .await + .expect("the second node answers correctly"); + + assert_eq!( + data.index, 0, + "the invalid answer must not be the one returned" + ); + } + /// A syncing node at the head of the list must not stop duties being /// refreshed when a healthy node sits behind it. /// diff --git a/crates/validator/src/beacon_node/http.rs b/crates/validator/src/beacon_node/http.rs index 013964c74..f32f39451 100644 --- a/crates/validator/src/beacon_node/http.rs +++ b/crates/validator/src/beacon_node/http.rs @@ -589,7 +589,7 @@ impl BeaconNodeApi for HttpBeaconNode { // is what lets `FallbackBeaconNode` move to the next node; the same // check one layer up runs only after this node's answer was already // accepted. - validate_attestation_data(slot, &data)?; + validate_attestation_data(slot, fork, &data)?; Ok(data) } diff --git a/crates/validator/src/beacon_node/mock.rs b/crates/validator/src/beacon_node/mock.rs index fe4a952b9..b5d46c907 100644 --- a/crates/validator/src/beacon_node/mock.rs +++ b/crates/validator/src/beacon_node/mock.rs @@ -599,7 +599,7 @@ impl BeaconNodeApi for MockBeaconNode { // head: it errors, and failover moves past it. Without this the mock // would be a more permissive node than any real one, and tests built // on it would not reflect what happens in production. - crate::beacon_node::validate_attestation_data(slot, &data)?; + crate::beacon_node::validate_attestation_data(slot, fork, &data)?; Ok(data) } diff --git a/crates/validator/src/beacon_node/mod.rs b/crates/validator/src/beacon_node/mod.rs index 49c1c54dd..5399d7ab3 100644 --- a/crates/validator/src/beacon_node/mod.rs +++ b/crates/validator/src/beacon_node/mod.rs @@ -32,8 +32,8 @@ pub mod http; #[cfg(test)] pub mod mock; -/// Check that `data` is an answer about `slot`, as the contract on -/// [`BeaconNodeApi::attestation_data`] requires. +/// Check that `data` is an answer about `slot`, which is in `fork`, as the +/// contract on [`BeaconNodeApi::attestation_data`] requires. /// /// One function rather than two copies, because it is enforced in two places /// for two different reasons and they must not drift apart. Each @@ -42,12 +42,21 @@ pub mod mock; /// before signing, since the trait is public and nothing stops an /// implementation from being wrong. /// -/// Both fields are checked because both are signable. `slot` is the obvious -/// one. `target.epoch` matters independently: it selects the signing domain -/// (see `SigningContext::attestation_signing_root`), so a node answering with -/// the right slot and a wrong target epoch is a second, distinct way to make -/// this client sign something it should not. -pub fn validate_attestation_data(slot: Slot, data: &AttestationData) -> Result<()> { +/// `slot` and `target.epoch` are checked because both are signable. `slot` is +/// the obvious one. `target.epoch` matters independently: it selects the +/// signing domain (see `SigningContext::attestation_signing_root`), so a node +/// answering with the right slot and a wrong target epoch is a second, +/// distinct way to make this client sign something it should not. +/// +/// `index` is checked for liveness rather than safety. A vote whose index +/// cannot be valid at `fork` is one every peer rejects (gossip's +/// `attestation.data.index == 0` from electra, `< 2` from gloas), and by +/// then slashing protection has recorded its target, so the validator has no +/// second vote to cast in that epoch. Refused here, nothing is recorded for +/// it, and the `Err` is one failover moves past to a node that may answer +/// properly. Lodestar's validator client refuses the same before signing +/// (`validateAttestationDuty`). +pub fn validate_attestation_data(slot: Slot, fork: ForkName, data: &AttestationData) -> Result<()> { if data.slot != slot { return Err(crate::error::Error::InconsistentResponse(format!( "requested attestation data for slot {slot}, node answered for slot {}", @@ -74,9 +83,51 @@ pub fn validate_attestation_data(slot: Slot, data: &AttestationData) -> Result<( data.source.epoch, data.target.epoch ))); } + let highest_index = highest_attestation_index(fork)?; + if data.index > highest_index { + return Err(crate::error::Error::InconsistentResponse(format!( + "attestation data for slot {slot} carries index {}, and a {} attestation's is at \ + most {highest_index}", + data.index, + fork.as_str() + ))); + } Ok(()) } +/// The highest `AttestationData.index` a vote at `fork` can carry. +/// +/// Every fork is named rather than compared against, so a fork added after +/// gloas has to say what its index means instead of inheriting gloas's +/// answer. +fn highest_attestation_index(fork: ForkName) -> Result { + match fork { + // A committee, and the one the query names: this client only ever + // asks about committee 0 (see `BeaconNodeApi::attestation_data`), so + // any other index is not an answer to its question. + ForkName::Phase0 + | ForkName::Altair + | ForkName::Bellatrix + | ForkName::Capella + | ForkName::Deneb => Ok(0), + // EIP-7549 moved the committee out of the signed data and left the + // field fixed at zero. + ForkName::Electra | ForkName::Fulu => Ok(0), + // The payload signal: 0 for a same-slot block or an empty one, 1 for + // a block whose payload was seen. Which of the two is right depends + // on the node's view of the block, which this client does not have, + // so only the range is checked. + ForkName::Gloas => Ok(1), + // Not a Beacon Chain fork: `Config::fork_at_epoch` never names it, so + // a caller that does is wrong, and nothing is signed for it. + ForkName::Lean => Err(crate::error::Error::InconsistentResponse( + "attestation data was asked for under the lean fork, which is not a Beacon Chain \ + fork and has no attestation of this shape" + .to_string(), + )), + } +} + /// The chain's genesis, as the beacon node reports it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Genesis { @@ -339,6 +390,12 @@ pub trait BeaconNodeApi: Send + Sync { /// path: a target from the future would raise the validator's watermark /// and cost it every attestation until the chain caught up. /// + /// It must also reject an `index` that cannot be valid at `fork`: anything + /// but 0 before gloas, anything above 1 from it. Such a vote is rejected + /// by every peer, after slashing protection has spent the validator's one + /// vote for the epoch on it. [`validate_attestation_data`] is every one of + /// these checks. + /// /// It matters that this is the *implementation's* job rather than the /// caller's, because of where failover sits. /// [`fallback::FallbackBeaconNode`] wraps this method: a check performed @@ -609,21 +666,74 @@ mod tests { /// Slot 100 is in epoch 3 on the mainnet preset. #[test] fn attestation_data_with_a_source_at_or_past_its_target_is_refused() { - assert!(validate_attestation_data(100, &data(100, 2, 3)).is_ok()); - assert!(validate_attestation_data(100, &data(100, 0, 3)).is_ok()); - let err = validate_attestation_data(100, &data(100, 3, 3)).expect_err("source == target"); + let fork = ForkName::Fulu; + assert!(validate_attestation_data(100, fork, &data(100, 2, 3)).is_ok()); + assert!(validate_attestation_data(100, fork, &data(100, 0, 3)).is_ok()); + let err = + validate_attestation_data(100, fork, &data(100, 3, 3)).expect_err("source == target"); assert!(err.to_string().contains("source epoch 3"), "{err}"); - assert!(validate_attestation_data(100, &data(100, 4, 3)).is_err()); + assert!(validate_attestation_data(100, fork, &data(100, 4, 3)).is_err()); } /// At genesis both checkpoints are epoch 0, and that is the one valid /// shape where the source does not precede the target. #[test] fn genesis_attestation_data_is_accepted() { - assert!(validate_attestation_data(0, &data(0, 0, 0)).is_ok()); + let fork = ForkName::Phase0; + assert!(validate_attestation_data(0, fork, &data(0, 0, 0)).is_ok()); assert!( - validate_attestation_data(0, &data(0, 1, 0)).is_err(), + validate_attestation_data(0, fork, &data(0, 1, 0)).is_err(), "a source past a genesis target is not the genesis shape" ); } + + fn with_index(index: u64) -> AttestationData { + AttestationData { + index, + ..data(100, 2, 3) + } + } + + /// Before gloas the index is zero: the committee this client asks about + /// up to deneb, and EIP-7549's fixed zero from electra. Anything else is + /// a vote no peer accepts. + #[test] + fn a_nonzero_index_is_refused_before_gloas() { + let before_gloas = ForkName::ALL + .iter() + .copied() + .filter(|fork| *fork < ForkName::Gloas); + for fork in before_gloas { + assert!( + validate_attestation_data(100, fork, &with_index(0)).is_ok(), + "{fork:?} takes index 0" + ); + let err = validate_attestation_data(100, fork, &with_index(1)) + .expect_err("an index of 1 before gloas"); + assert!( + matches!(err, crate::error::Error::InconsistentResponse(_)), + "{fork:?}: failover must be able to move past it, got {err}" + ); + assert!(err.to_string().contains("carries index 1"), "{err}"); + } + } + + /// From gloas the index is the payload signal, 0 or 1, and passes through + /// as the node gave it. Only a value outside that range is refused. + #[test] + fn a_gloas_index_is_zero_or_one() { + let fork = ForkName::Gloas; + assert!(validate_attestation_data(100, fork, &with_index(0)).is_ok()); + assert!(validate_attestation_data(100, fork, &with_index(1)).is_ok()); + let err = validate_attestation_data(100, fork, &with_index(2)).expect_err("an index of 2"); + assert!(err.to_string().contains("at most 1"), "{err}"); + assert!(validate_attestation_data(100, fork, &with_index(u64::MAX)).is_err()); + } + + /// Lean is not on the Beacon Chain's fork timeline, so no index is valid + /// for it and nothing is signed. + #[test] + fn the_lean_fork_has_no_valid_index() { + assert!(validate_attestation_data(100, ForkName::Lean, &with_index(0)).is_err()); + } } diff --git a/docs/cli.md b/docs/cli.md index 5f75850f5..d4575c661 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -311,8 +311,12 @@ advance so it can collect the votes. Each slot it wakes at the boundary. If one of its validators proposes that slot, it signs the RANDAO reveal, asks the beacon node for a block, checks that the block is for the slot and proposer it asked about, signs it and publishes -it. One third into the slot it fetches the attestation data, signs for every -validator due that slot, and submits the batch. Two thirds in, for any duty it +it. One third into the slot it fetches the attestation data, checks that it is +for the slot it asked about and that its `index` is one the slot's fork allows +(0 before gloas, 0 or 1 from it, where it is the payload signal), signs for +every validator due that slot, and submits the batch. A node whose answer fails +either check is passed over for the next one in `--beacon-nodes`, before +anything is recorded or signed. Two thirds in, for any duty it was selected to aggregate, it asks the node for the aggregate covering that committee's votes on the data it just signed, wraps it with the selection proof and publishes it. From 4f2d9b166ce620ba561dfdf78f4c0d2e6fe17b9f Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Wed, 11 Nov 2026 17:58:27 -0300 Subject: [PATCH 3/6] Stop the validator client on a poisoned slashing-protection record Once a panic poisoned the record's lock, every duty was refused until a restart, but the process stayed up and answered /health, so nothing that supervises it had a reason to restart it. The duty loop now asks slashing_protection::ensure_usable at each slot boundary and returns the error from run on a poisoned lock: the process exits with a failure, and systemd or Kubernetes applies the remedy the refusals ask for. The restart rereads the record from disk, which every write either replaced whole or left alone. The access whose own operation panicked was told "an earlier access panicked", which points at an event that never happened. It is now told it was its own panic, with the panic's message; only later accesses get the old wording. docs/cli.md says to run the client under a supervisor. --- crates/validator/src/error.rs | 12 +- crates/validator/src/lib.rs | 12 ++ .../validator/src/slashing_protection/mod.rs | 134 +++++++++++++++--- docs/cli.md | 7 +- 4 files changed, 143 insertions(+), 22 deletions(-) diff --git a/crates/validator/src/error.rs b/crates/validator/src/error.rs index ae65aee92..0aacfe313 100644 --- a/crates/validator/src/error.rs +++ b/crates/validator/src/error.rs @@ -115,15 +115,17 @@ pub enum Error { #[error("slashing protection {path}: {reason}")] SlashingProtection { path: String, reason: String }, /// The record cannot be reached at all, as opposed to a record that - /// answered: its lock is poisoned because an earlier access panicked - /// while holding it, or the runtime is shutting down under the blocking - /// task. + /// answered: its lock is poisoned because an access panicked while + /// holding it, it stayed held past the wait, or the runtime is shutting + /// down under the blocking task. /// /// Kept apart from `SlashingProtection` and `Io` because the remedy /// differs: those name a file to look at, this one names a process to /// restart. A poisoned lock is a bug, never a disk, and it stays poisoned - /// for the life of the process, so every duty is refused until it is - /// restarted. Never retryable. + /// for the life of the process, so every duty is refused from then on + /// and the duty loop stops the client at the next slot, for its + /// supervisor to restart (see `slashing_protection::ensure_usable`). + /// Never retryable. #[error("slashing protection is unusable: {reason}")] SlashingProtectionUnusable { reason: String }, diff --git a/crates/validator/src/lib.rs b/crates/validator/src/lib.rs index 24fc99dfd..4b583c16a 100644 --- a/crates/validator/src/lib.rs +++ b/crates/validator/src/lib.rs @@ -265,6 +265,18 @@ pub async fn run(config: ValidatorConfig) -> Result<()> { tokio::time::sleep(delay).await; served = Some(slot); + // A poisoned record refuses every duty until the process restarts, + // and nothing in the process restarts it. Carrying on would leave a + // client that answers `/health` and signs nothing, which no + // supervisor has a reason to touch; stopping hands systemd or + // Kubernetes the restart the refusals ask for. Asked once a slot, at + // the boundary, which is soon enough: in between, the record refuses + // every duty on its own. + if let Err(err) = slashing_protection::ensure_usable(&protection) { + error!(%slot, %err, "Slashing protection is unusable; stopping so the client can be restarted"); + return Err(err); + } + let epoch = clock.epoch_of(slot); if last_epoch != Some(epoch) { diff --git a/crates/validator/src/slashing_protection/mod.rs b/crates/validator/src/slashing_protection/mod.rs index 909d14e32..d462d490e 100644 --- a/crates/validator/src/slashing_protection/mod.rs +++ b/crates/validator/src/slashing_protection/mod.rs @@ -96,9 +96,11 @@ const RECORD_POLL: std::time::Duration = std::time::Duration::from_millis(5); /// could not be reached at all (`Error::SlashingProtectionUnusable`): the /// lock is poisoned, because an operation panicked while holding it, and /// stays so until the process restarts, since nothing here recovers a record -/// that a panic may have left half-updated in memory. The panic that poisons -/// it is reported the same way on the call that raised it, which sees it as -/// the blocking task's `JoinError`. +/// that a panic may have left half-updated in memory. The call whose own +/// operation panicked sees it as the blocking task's `JoinError` and is told +/// so, with the panic's message; every later call is told an earlier access +/// panicked. The duty loop asks [`ensure_usable`] once a slot and stops the +/// client on a poisoned record, so the restart comes from its supervisor. /// /// The operation runs to completion even if the caller is dropped, as a duty /// abandoned at its deadline is: that is safe, since a record written for a @@ -113,11 +115,6 @@ pub async fn blocking( protection: &SharedProtection, operation: impl FnOnce(&mut SlashingProtection) -> Result + Send + 'static, ) -> Result { - let poisoned = || Error::SlashingProtectionUnusable { - reason: "an earlier access panicked while holding the record, so nothing is signed \ - until the validator client is restarted" - .to_string(), - }; let protection = protection.clone(); tokio::task::spawn_blocking(move || { // A bounded wait rather than `lock()`: a write normally holds the @@ -129,7 +126,7 @@ pub async fn blocking( let mut protection = loop { match protection.try_lock() { Ok(guard) => break guard, - Err(std::sync::TryLockError::Poisoned(_)) => return Err(poisoned()), + Err(std::sync::TryLockError::Poisoned(_)) => return Err(poisoned_earlier()), Err(std::sync::TryLockError::WouldBlock) => { if std::time::Instant::now() >= deadline { return Err(Error::SlashingProtectionUnusable { @@ -148,16 +145,58 @@ pub async fn blocking( }) .await .map_err(|err| { - if err.is_panic() { - poisoned() - } else { - Error::SlashingProtectionUnusable { - reason: format!("the record could not be consulted: {err}"), - } - } + let reason = match err.try_into_panic() { + // This call's own operation panicked, and it held the record when + // it did: nothing else in the task can panic. Said as this + // access's failure, with the panic's message, since "an earlier + // access" would send whoever reads it looking for another event. + Ok(panic) => format!( + "this access panicked while holding the record ({}), so nothing is signed until \ + the validator client is restarted", + panic_message(panic.as_ref()) + ), + Err(err) => format!("the record could not be consulted: {err}"), + }; + Error::SlashingProtectionUnusable { reason } })? } +/// What a caller is told when it finds the record's lock already poisoned. +fn poisoned_earlier() -> Error { + Error::SlashingProtectionUnusable { + reason: "an earlier access panicked while holding the record, so nothing is signed \ + until the validator client is restarted" + .to_string(), + } +} + +/// A panic's message, when it carried one: `panic!` with a literal gives a +/// `&str` and with arguments a `String`. Anything else was raised with +/// `panic_any` and has no text to show. +fn panic_message(panic: &(dyn std::any::Any + Send)) -> &str { + panic + .downcast_ref::<&'static str>() + .copied() + .or_else(|| panic.downcast_ref::().map(String::as_str)) + .unwrap_or("no message") +} + +/// Refuse to go on with a record whose lock is poisoned. +/// +/// A poisoned record refuses every duty and never recovers (see +/// [`blocking`]), so a client left running on one looks alive to whatever +/// supervises it while signing nothing. The duty loop asks this once a slot +/// and stops on an `Err`, which ends the process with a failure status: +/// systemd or Kubernetes then restarts it, and the restart reads the record +/// back from disk, which every write either replaced whole or left alone. +/// That is the remedy the refusals ask for. +pub fn ensure_usable(protection: &SharedProtection) -> Result<()> { + if protection.is_poisoned() { + return Err(poisoned_earlier()); + } + Ok(()) +} + /// Whether an attestation's source epoch is at or past its target, which is /// never valid outside genesis: with unsigned epochs, `(0, 0)` is the only /// such pair that is. One predicate for the three places that refuse it, @@ -1542,4 +1581,67 @@ mod tests { Err(Refusal::Unregistered) ); } + + // A poisoned record. + + /// The duty loop's once-a-slot question: a healthy record lets it go on, + /// a poisoned one stops the client, and the error is not one to retry. + #[test] + fn a_poisoned_record_stops_the_client() { + let protection = SlashingProtection::in_memory(); + ensure_usable(&protection).expect("a healthy record is usable"); + + // Poisoned the way a bug would: a panic while the lock is held. + let held = protection.clone(); + std::thread::spawn(move || { + let _guard = held.lock().expect("not poisoned yet"); + panic!("a bug while holding the record"); + }) + .join() + .expect_err("the thread panics"); + + let err = ensure_usable(&protection).expect_err("a poisoned record is not"); + assert!( + matches!(err, Error::SlashingProtectionUnusable { .. }), + "got {err}" + ); + assert!(!err.is_retryable(), "the loop must stop, not try again"); + } + + /// The access whose own operation panics is told it was its panic, with + /// the message, and only the ones after it that an earlier access + /// panicked: the same wording for both points the first at an event that + /// never happened. + #[tokio::test] + async fn the_access_that_panics_is_told_so_and_later_ones_that_an_earlier_one_did() { + let protection = SlashingProtection::in_memory(); + + let own = blocking(&protection, |_| -> Result<()> { + panic!("a bug in the record") + }) + .await + .expect_err("the operation panicked") + .to_string(); + assert!(own.contains("this access panicked"), "{own}"); + assert!(own.contains("a bug in the record"), "{own}"); + + let later = blocking(&protection, |_| Ok(())) + .await + .expect_err("the record is poisoned") + .to_string(); + assert!(later.contains("an earlier access panicked"), "{later}"); + ensure_usable(&protection).expect_err("and the duty loop stops on it"); + } + + /// A panic raised with arguments carries a `String`, not a `&str`, and + /// one raised with `panic_any` carries neither. + #[test] + fn a_panics_message_is_read_whatever_it_was_raised_with() { + let literal: Box = Box::new("a literal"); + assert_eq!(panic_message(literal.as_ref()), "a literal"); + let formatted: Box = Box::new(format!("slot {}", 7)); + assert_eq!(panic_message(formatted.as_ref()), "slot 7"); + let neither: Box = Box::new(7_u64); + assert_eq!(panic_message(neither.as_ref()), "no message"); + } } diff --git a/docs/cli.md b/docs/cli.md index d4575c661..172d96c8b 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -277,7 +277,12 @@ implementation's, so none of the [common flags](#common-flags) apply to it. > regular file, since a dangling one would read as "no record"; and > the directory should be a local disk, since every signature waits for the > record to be flushed to it, and the lock that keeps a second process off it -> is an `flock`, which a network filesystem may not honour. Unix only. A key with no record is refused at startup unless +> is an `flock`, which a network filesystem may not honour. Unix only. Run the +> client under a supervisor that restarts it (systemd's `Restart=on-failure`, +> a Kubernetes pod): if a bug ever panics while the record is held, the record +> is unusable for the rest of that process, every duty is refused, and the +> client exits with a failure at the next slot instead of idling, so that the +> restart rereads the record from disk. A key with no record is refused at startup unless > `--init-slashing-protection` is passed. A key moving here from another client > brings its history with [`slashing-protection import`](#slashing-protection), > or in the same keymanager API request that imports the key. See From 11bec9b5a21ad44629a0511977983d38cc7a8d3c Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Thu, 12 Nov 2026 17:30:47 -0300 Subject: [PATCH 4/6] Stop on a poisoned record at once, and let the keymanager finish first The duty loop asked ensure_usable once a slot, so a poisoned record was seen up to a slot late, and run returned with the keymanager server still serving. An import caught between saving its keys and registering them was cut off there, leaving keys on disk with no record, which every later start refuses. SharedProtection is now a handle carrying the record and a signal. The blocking task raises the signal itself: when it finds the lock poisoned, and, through a drop guard, when its own operation panics, so it is raised even if the caller gave up at its deadline. run races the duty loop against it and the once-a-slot poll is gone. Whatever ends the loop, run then tells the keymanager server to stop accepting requests and waits for the ones in flight (http_api::serve, axum's graceful shutdown). The import above meets the poisoned record, takes its keys off disk again and answers. The wait is bounded at 60 s, so a request that never ends cannot hold the restart up. The access whose operation panicked is still told the panic was its own, but no longer repeats the panic's message: the panic hook prints it, with its location, just before. That removes a second copy of the payload downcast. Only attestations and proposals consult the record, so the docs no longer say a poisoned one refuses every duty. tokio-util joins the validator crate for CancellationToken. --- Cargo.lock | 1 + crates/validator/Cargo.toml | 4 + crates/validator/src/attestation.rs | 2 +- crates/validator/src/error.rs | 9 +- crates/validator/src/http_api/keystores.rs | 126 +++++- crates/validator/src/http_api/mod.rs | 20 + crates/validator/src/lib.rs | 407 +++++++++++++----- crates/validator/src/proposal.rs | 4 +- .../validator/src/slashing_protection/mod.rs | 238 ++++++---- docs/cli.md | 7 +- 10 files changed, 599 insertions(+), 219 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..18491a5c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2290,6 +2290,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tokio-util", "tower", "tracing", "unicode-normalization", diff --git a/crates/validator/Cargo.toml b/crates/validator/Cargo.toml index b34e72488..cc2bed677 100644 --- a/crates/validator/Cargo.toml +++ b/crates/validator/Cargo.toml @@ -31,6 +31,10 @@ tracing.workspace = true # Declared here rather than relied on through workspace feature unification, # so this crate still builds if it is ever compiled on its own. tokio = { workspace = true, features = ["time", "macros", "rt"] } +# `CancellationToken`: the signal that stops the client when the +# slashing-protection record becomes unusable, and the one that tells the +# keymanager server to finish its requests in flight. +tokio-util.workspace = true reqwest = { workspace = true, features = ["json"] } # BLS12-381. Same version and backend the state transition verifies with, so a diff --git a/crates/validator/src/attestation.rs b/crates/validator/src/attestation.rs index 9de0b770b..8e773610f 100644 --- a/crates/validator/src/attestation.rs +++ b/crates/validator/src/attestation.rs @@ -427,7 +427,7 @@ mod tests { keys: &[ethlambda_types::beacon::primitives::BlsPubkey], ) -> SharedProtection { let protection = SlashingProtection::open(dir, Root::ZERO, keys, true).expect("opens"); - Arc::new(std::sync::Mutex::new(protection)) + crate::slashing_protection::SharedProtection::new(protection) } /// The restart shape the old in-memory guard could not cover: a fresh diff --git a/crates/validator/src/error.rs b/crates/validator/src/error.rs index 0aacfe313..33ec4f3d5 100644 --- a/crates/validator/src/error.rs +++ b/crates/validator/src/error.rs @@ -122,10 +122,11 @@ pub enum Error { /// Kept apart from `SlashingProtection` and `Io` because the remedy /// differs: those name a file to look at, this one names a process to /// restart. A poisoned lock is a bug, never a disk, and it stays poisoned - /// for the life of the process, so every duty is refused from then on - /// and the duty loop stops the client at the next slot, for its - /// supervisor to restart (see `slashing_protection::ensure_usable`). - /// Never retryable. + /// for the life of the process. Only attestations and proposals consult + /// the record, so those are what it refuses from then on; the other + /// duties would go on signing, which is why the client stops as soon as + /// the poisoning is seen, for its supervisor to restart (see + /// `slashing_protection::SharedProtection`). Never retryable. #[error("slashing protection is unusable: {reason}")] SlashingProtectionUnusable { reason: String }, diff --git a/crates/validator/src/http_api/keystores.rs b/crates/validator/src/http_api/keystores.rs index 72dbbe77e..1dce5dd7a 100644 --- a/crates/validator/src/http_api/keystores.rs +++ b/crates/validator/src/http_api/keystores.rs @@ -1376,7 +1376,7 @@ mod tests { .expect("opens with no keys"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let protection = context.protection.clone(); let (_, body) = send( @@ -1420,7 +1420,7 @@ mod tests { std::fs::create_dir(&record).expect("a directory in its place"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let store = context.store.clone(); let (_, body) = send( @@ -1493,7 +1493,7 @@ mod tests { .expect("saves"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let (_, body) = send( router(context, TOKEN.to_string()), @@ -1563,7 +1563,7 @@ mod tests { .expect("saves"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let (_, body) = send( router(context, TOKEN.to_string()), @@ -1694,7 +1694,7 @@ mod tests { std::fs::create_dir(&record).expect("a directory in its place"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let (_, body) = send( router(context, TOKEN.to_string()), @@ -1732,6 +1732,118 @@ mod tests { ); } + /// The reason the client stops gracefully. An import has saved its key + /// and is waiting to register it when a bug poisons the record: the + /// client stops, which used to drop the runtime with the request parked + /// there, leaving a key saved and enabled with no record, refused at + /// every start after. Told to stop instead, the server lets the import + /// meet the poisoned record, take its key off disk again and answer. + #[tokio::test] + async fn an_import_in_flight_when_the_record_is_poisoned_is_undone_before_the_client_stops() { + let dir = tempfile::tempdir().expect("temp dir"); + let (context, protection) = context_with_record(dir.path()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("binds"); + let address = listener.local_addr().expect("bound"); + let shutdown = tokio_util::sync::CancellationToken::new(); + let server = tokio::spawn({ + let router = router(context, TOKEN.to_string()); + let shutdown = shutdown.clone(); + async move { + crate::http_api::serve(listener, router, shutdown) + .await + .expect("serves"); + } + }); + + // Another access holds the record, so the import's registration + // waits for it, and will panic with it held: the bug. + let (release, held) = std::sync::mpsc::channel::<()>(); + let (locked, is_locked) = std::sync::mpsc::channel::<()>(); + let holder = std::thread::spawn({ + let protection = protection.clone(); + move || { + let _guard = protection.lock().expect("not poisoned yet"); + locked.send(()).expect("the test is waiting"); + held.recv().expect("released"); + panic!("a bug while holding the record"); + } + }); + is_locked.recv().expect("the record is held"); + + let request = tokio::spawn(async move { + reqwest::Client::new() + .post(format!("http://{address}/eth/v1/keystores")) + .bearer_auth(TOKEN) + .json(&serde_json::json!({ + "keystores": [keystore_json()], + "passwords": [PASSWORD], + })) + .send() + .await + .expect("the request in flight is answered") + .json::() + .await + .expect("with a body") + }); + + // The import has saved its key once the definitions file lists it. + let saved = async { + loop { + let listed = ValidatorDefinitions::open(dir.path()) + .map(|definitions| !definitions.0.is_empty()) + .unwrap_or(false); + if listed { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }; + tokio::time::timeout(std::time::Duration::from_secs(60), saved) + .await + .expect("the import saves its key"); + release.send(()).expect("the holder is waiting"); + holder.join().expect_err("the holder panics"); + + // What `run` does from here: stop on the record, the keymanager last. + let stopped = tokio::time::timeout( + std::time::Duration::from_secs(60), + crate::run_until_stopped(std::future::pending(), &protection, shutdown, Some(server)), + ) + .await + .expect("the client stops"); + assert!( + matches!(stopped, Error::SlashingProtectionUnusable { .. }), + "got {stopped}" + ); + + // Checked before the answer is read: by the time the client is + // reported stopped, which is when the runtime goes, the import has + // already been undone. + assert!( + ValidatorStore::load(dir.path()) + .expect("loads") + .pubkeys() + .is_empty(), + "the next start finds no key to refuse" + ); + assert!( + !dir.path().join(format!("{KEYSTORE_PUBKEY}.json")).exists(), + "the keystore is off disk again" + ); + + let body = request.await.expect("the request task ends"); + assert_eq!(body["data"][0]["status"], "error", "got {body}"); + assert!( + body["data"][0]["message"] + .as_str() + .unwrap_or_default() + .starts_with("not imported"), + "got {body}" + ); + } + /// A key that fails to persist must leave no slashing-protection record /// behind: a record would later let the same keystore, copied in by hand, /// past the startup check as if it had been registered on purpose. @@ -1752,7 +1864,7 @@ mod tests { .expect("opens with no keys"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let protection = context.protection.clone(); let (_, body) = send( @@ -1789,7 +1901,7 @@ mod tests { ) .expect("opens with no keys"); let mut context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir); - context.protection = Arc::new(std::sync::Mutex::new(protection)); + context.protection = crate::slashing_protection::SharedProtection::new(protection); let protection = context.protection.clone(); (context, protection) } diff --git a/crates/validator/src/http_api/mod.rs b/crates/validator/src/http_api/mod.rs index 189e5ac7a..e21b01faf 100644 --- a/crates/validator/src/http_api/mod.rs +++ b/crates/validator/src/http_api/mod.rs @@ -16,6 +16,7 @@ use axum::routing::get; use axum::{Extension, Router}; use subtle::ConstantTimeEq as _; use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; use tracing::info; use crate::Error; @@ -173,6 +174,25 @@ pub fn router(context: KeymanagerContext, token: String) -> Router { /// The largest request body the keymanager API reads. const KEYMANAGER_BODY_LIMIT: usize = 64 * 1024 * 1024; +/// Serve the keymanager API on `listener` until `shutdown` is cancelled, then +/// stop accepting requests and return once the ones in flight have answered. +/// +/// Stopping that way matters to an import. It saves its keys before it +/// registers them, and takes them off disk again if the registration fails +/// (see [`keystores::import`]). Dropped in between, as it would be if the +/// runtime went away under it, it leaves keys that are saved and enabled with +/// no slashing-protection record, which the next start refuses, and the start +/// after that. +pub async fn serve( + listener: tokio::net::TcpListener, + router: Router, + shutdown: CancellationToken, +) -> std::io::Result<()> { + axum::serve(listener, router) + .with_graceful_shutdown(shutdown.cancelled_owned()) + .await +} + #[derive(Clone)] struct ApiToken(String); diff --git a/crates/validator/src/lib.rs b/crates/validator/src/lib.rs index 4b583c16a..7c5db7724 100644 --- a/crates/validator/src/lib.rs +++ b/crates/validator/src/lib.rs @@ -51,6 +51,8 @@ use std::time::SystemTime; use ethlambda_types::beacon::primitives::{BlsPubkey, Bytes32, ExecutionAddress, ValidatorIndex}; use tokio::sync::{Mutex, RwLock}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use crate::aggregation::AggregationService; @@ -196,7 +198,7 @@ pub async fn run(config: ValidatorConfig) -> Result<()> { path = %protection.path().map(|path| path.display().to_string()).unwrap_or_default(), "Slashing protection loaded" ); - let protection = Arc::new(std::sync::Mutex::new(protection)); + let protection = slashing_protection::SharedProtection::new(protection); let context = Arc::new(SigningContext { config: spec, @@ -216,6 +218,11 @@ pub async fn run(config: ValidatorConfig) -> Result<()> { protection.clone(), ); + // Cancelled when the duty loop below ends, whatever ended it: the + // keymanager server then stops accepting requests and finishes the ones + // in flight, and `run` waits for it before returning. + let shutdown = CancellationToken::new(); + let mut keymanager = None; if let Some(address) = config.keymanager { let token = http_api::load_or_create_token(&config.validators_dir)?; let context = http_api::KeymanagerContext { @@ -235,129 +242,193 @@ pub async fn run(config: ValidatorConfig) -> Result<()> { source, })?; info!(%address, "Keymanager API listening"); - tokio::spawn(async move { - if let Err(err) = axum::serve(listener, router).await { + let shutdown = shutdown.clone(); + keymanager = Some(tokio::spawn(async move { + if let Err(err) = http_api::serve(listener, router, shutdown).await { error!(%err, "Keymanager API stopped"); } - }); + })); } - let mut last_epoch: Option = None; - // The slot whose duties were last attempted, so an overrunning slot does - // not cost the next one as well. See `SlotClock::next_slot_to_serve`. - let mut served: Option = None; - - loop { - // One wake per slot, at the slot boundary, rather than one at the - // attester offset. - // - // A slot has more than one thing due in it and they are due at - // different points: a proposer publishes at the boundary, every - // attester votes one third in. A loop woken only at the attester - // offset can never propose, because by the time it runs the slot it - // would be proposing for is already a third gone. - // - // Waking at the boundary and sleeping further in reaches both, and - // gives the epoch refresh below a third of a slot of headroom it did - // not have when it ran at the attester offset itself, where every - // second it took came straight out of the attestation's lateness. - let (slot, delay) = clock.next_slot_to_serve(served, SystemTime::now()); - tokio::time::sleep(delay).await; - served = Some(slot); - - // A poisoned record refuses every duty until the process restarts, - // and nothing in the process restarts it. Carrying on would leave a - // client that answers `/health` and signs nothing, which no - // supervisor has a reason to touch; stopping hands systemd or - // Kubernetes the restart the refusals ask for. Asked once a slot, at - // the boundary, which is soon enough: in between, the record refuses - // every duty on its own. - if let Err(err) = slashing_protection::ensure_usable(&protection) { - error!(%slot, %err, "Slashing protection is unusable; stopping so the client can be restarted"); - return Err(err); - } - - let epoch = clock.epoch_of(slot); - - if last_epoch != Some(epoch) { - // Snapshot the pubkeys and let the guard drop right here, before - // `refresh_epoch`'s awaits (a sync check, two duties fetches, a - // validator lookup, a subscription POST). Passing the guard - // itself in used to extend it across that whole call, because - // `&*store.read().await` as a match scrutinee has its temporary - // lifetime extended to the entire match. Holding a read lock - // that long lets the keymanager's write lock - // (`http_api::keystores::import`, held across slow EIP-2335 - // derivation) queue ahead of it, and since `tokio::sync::RwLock` - // is write-preferring, every read after that point, including - // this one next epoch, would starve until the import finished. - let pubkeys = store.read().await.pubkeys(); - let refreshed = refresh_epoch( - &beacon_node, - &mut duties, - &pubkeys, - epoch, - &settings, - &store, - &context, - ); - match refreshed.await { - Ok(()) => last_epoch = Some(epoch), - // A beacon node that is down, syncing or answering badly will - // likely answer next epoch, so keep the schedule already held - // and try again. Anything else will not fix itself, and looping - // on it would hide the cause behind a warning every slot. - Err(err) if err.is_retryable() => { - warn!(%epoch, %err, "Failed to refresh duties; keeping the previous schedule"); - metrics::set_beacon_node_available(false); - } - // Unreachable today: every error `refresh_epoch` can produce - // comes from a `BeaconNodeApi` call, and every variant those - // raise is classified retryable (see `Error::is_retryable`). - // Kept rather than deleted, since it is the only thing that - // would stop the process if a future non-retryable failure - // (a local config or key problem, say) ever reached this - // point; do not remove it as unreachable dead code. - Err(err) => { - error!(%epoch, %err, "Cannot refresh duties; stopping"); - return Err(err); + // The duty loop, as a future that ends only with the error that stopped + // it. Not awaited here: `run_until_stopped` races it against the + // slashing-protection record becoming unusable. + let duty_loop = async { + let mut last_epoch: Option = None; + // The slot whose duties were last attempted, so an overrunning slot does + // not cost the next one as well. See `SlotClock::next_slot_to_serve`. + let mut served: Option = None; + + loop { + // One wake per slot, at the slot boundary, rather than one at the + // attester offset. + // + // A slot has more than one thing due in it and they are due at + // different points: a proposer publishes at the boundary, every + // attester votes one third in. A loop woken only at the attester + // offset can never propose, because by the time it runs the slot it + // would be proposing for is already a third gone. + // + // Waking at the boundary and sleeping further in reaches both, and + // gives the epoch refresh below a third of a slot of headroom it did + // not have when it ran at the attester offset itself, where every + // second it took came straight out of the attestation's lateness. + let (slot, delay) = clock.next_slot_to_serve(served, SystemTime::now()); + tokio::time::sleep(delay).await; + served = Some(slot); + + let epoch = clock.epoch_of(slot); + + if last_epoch != Some(epoch) { + // Snapshot the pubkeys and let the guard drop right here, before + // `refresh_epoch`'s awaits (a sync check, two duties fetches, a + // validator lookup, a subscription POST). Passing the guard + // itself in used to extend it across that whole call, because + // `&*store.read().await` as a match scrutinee has its temporary + // lifetime extended to the entire match. Holding a read lock + // that long lets the keymanager's write lock + // (`http_api::keystores::import`, held across slow EIP-2335 + // derivation) queue ahead of it, and since `tokio::sync::RwLock` + // is write-preferring, every read after that point, including + // this one next epoch, would starve until the import finished. + let pubkeys = store.read().await.pubkeys(); + let refreshed = refresh_epoch( + &beacon_node, + &mut duties, + &pubkeys, + epoch, + &settings, + &store, + &context, + ); + match refreshed.await { + Ok(()) => last_epoch = Some(epoch), + // A beacon node that is down, syncing or answering badly will + // likely answer next epoch, so keep the schedule already held + // and try again. Anything else will not fix itself, and looping + // on it would hide the cause behind a warning every slot. + Err(err) if err.is_retryable() => { + warn!(%epoch, %err, "Failed to refresh duties; keeping the previous schedule"); + metrics::set_beacon_node_available(false); + } + // Unreachable today: every error `refresh_epoch` can produce + // comes from a `BeaconNodeApi` call, and every variant those + // raise is classified retryable (see `Error::is_retryable`). + // Kept rather than deleted, since it is the only thing that + // would stop the process if a future non-retryable failure + // (a local config or key problem, say) ever reached this + // point; do not remove it as unreachable dead code. + Err(err) => { + error!(%epoch, %err, "Cannot refresh duties; stopping"); + return err; + } } } - } - // A block is due now, at the boundary this iteration woke on. - // - // Cloned out of `duties` so the borrow ends before the await: the duty - // is three small fields, and holding a borrow of the schedule across - // the proposal would stop the next epoch's refresh from replacing it. - if let Some(duty) = duties.proposer_at_slot(slot, epoch).cloned() { - propose(&proposal, &clock, slot, &duty, &store).await; + // A block is due now, at the boundary this iteration woke on. + // + // Cloned out of `duties` so the borrow ends before the await: the duty + // is three small fields, and holding a borrow of the schedule across + // the proposal would stop the next epoch's refresh from replacing it. + if let Some(duty) = duties.proposer_at_slot(slot, epoch).cloned() { + propose(&proposal, &clock, slot, &duty, &store).await; + } + + // The rest of the way to the first of the slot's remaining duties: + // the attester offset, or the sync committee offset when the network + // puts it earlier. Zero if the refresh or the proposal above already + // ran past it, in which case the duty is late rather than skipped. + // Each branch of `serve_slot_duties` waits for its own offset. + tokio::time::sleep(clock.until_first_slot_duty(slot, SystemTime::now())).await; + + let slot_duties = duties.at_slot(slot, epoch); + let committee = duties.ptc_at_slot(slot, epoch); + let sync_duties = duties.sync_at_slot(slot); + serve_slot_duties( + &attestation, + &aggregation, + &payload_attestation, + &sync_committee, + &clock, + slot, + &slot_duties, + &committee, + &sync_duties, + &store, + ) + .await; } + }; - // The rest of the way to the first of the slot's remaining duties: - // the attester offset, or the sync committee offset when the network - // puts it earlier. Zero if the refresh or the proposal above already - // ran past it, in which case the duty is late rather than skipped. - // Each branch of `serve_slot_duties` waits for its own offset. - tokio::time::sleep(clock.until_first_slot_duty(slot, SystemTime::now())).await; + Err(run_until_stopped(duty_loop, &protection, shutdown, keymanager).await) +} - let slot_duties = duties.at_slot(slot, epoch); - let committee = duties.ptc_at_slot(slot, epoch); - let sync_duties = duties.sync_at_slot(slot); - serve_slot_duties( - &attestation, - &aggregation, - &payload_attestation, - &sync_committee, - &clock, - slot, - &slot_duties, - &committee, - &sync_duties, - &store, - ) - .await; +/// How long the keymanager gets to finish its requests in flight once the +/// client is stopping. +/// +/// Generous on purpose. An import decrypts every keystore of its batch before +/// it registers any, tens of seconds for a full one, and cut short between +/// saving its keys and registering them it leaves keys the next start +/// refuses. Nothing waits this long unless a request really is in flight, and +/// one that never ends must not keep a stopped client from being restarted. +const KEYMANAGER_DRAIN: std::time::Duration = std::time::Duration::from_secs(60); + +/// Run `duties` until they stop on their own or the slashing-protection +/// record becomes unusable, then let the keymanager finish, and say why the +/// client stopped. +/// +/// # Stopping on an unusable record +/// +/// A poisoned record signs no attestation and no block for the rest of the +/// process (see [`slashing_protection::SharedProtection`]). Carrying on would +/// leave a client that answers `/health` and does not attest, which nothing +/// that supervises it has a reason to touch; stopping hands systemd or +/// Kubernetes the restart. The duties are dropped where they stand, at once, +/// since the access that poisons the record is the one that raises the +/// signal. Dropping them is safe for the reason abandoning one at its +/// deadline is: whatever was signed was recorded first. +/// +/// # The keymanager goes last +/// +/// Returning from [`run`] drops the runtime, and a request dropped with it +/// stops at whichever await it was parked on. For an import that can be +/// between saving its keys and registering them, the one place it must not +/// stop (see [`http_api::serve`]). So the server is told to stop and waited +/// for: on a poisoned record the import's registration fails, and it takes +/// its keys off disk again before it answers. +async fn run_until_stopped( + duties: impl Future, + protection: &slashing_protection::SharedProtection, + shutdown: CancellationToken, + keymanager: Option>, +) -> Error { + let stopped = tokio::select! { + err = duties => err, + () = protection.unusable() => { + let err = Error::SlashingProtectionUnusable { + reason: "an access panicked while holding the record, so no attestation or \ + block is signed until the validator client is restarted" + .to_string(), + }; + error!(%err, "Stopping so the validator client can be restarted"); + err + } + }; + + shutdown.cancel(); + if let Some(keymanager) = keymanager { + match tokio::time::timeout(KEYMANAGER_DRAIN, keymanager).await { + Ok(joined) => { + let _ = + joined.inspect_err(|err| error!(%err, "Keymanager API failed while stopping")); + } + Err(_) => warn!( + waited_s = KEYMANAGER_DRAIN.as_secs(), + "Keymanager API still has a request in flight; stopping without it" + ), + } } + stopped } /// Everything a slot owes after its proposal: attest and aggregate, then the @@ -1423,4 +1494,114 @@ mod tests { .await; assert_eq!(node.payload_attestation_data_call_count(), 0); } + + // Stopping. + + /// Stands in for the keymanager server with a request in flight: told to + /// stop, it takes `work` to finish, and the flag says when it has. + fn keymanager_finishing_in( + shutdown: &CancellationToken, + work: std::time::Duration, + ) -> (JoinHandle<()>, Arc) { + let finished = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let handle = tokio::spawn({ + let shutdown = shutdown.clone(); + let finished = finished.clone(); + async move { + shutdown.cancelled().await; + tokio::time::sleep(work).await; + finished.store(true, std::sync::atomic::Ordering::Release); + } + }); + (handle, finished) + } + + /// A record poisoned by one of its accesses stops duties that would + /// otherwise never end, and the client is not reported stopped until the + /// keymanager has finished what it had in flight. + #[tokio::test(start_paused = true)] + async fn an_unusable_record_stops_the_client_once_the_keymanager_has_finished() { + let protection = slashing_protection::SlashingProtection::in_memory(); + let shutdown = CancellationToken::new(); + let (keymanager, finished) = + keymanager_finishing_in(&shutdown, std::time::Duration::from_secs(5)); + + slashing_protection::blocking(&protection, |_| -> Result<()> { + panic!("a bug in the record") + }) + .await + .expect_err("the access panicked"); + + let stopped = run_until_stopped( + std::future::pending(), + &protection, + shutdown.clone(), + Some(keymanager), + ) + .await; + assert!( + matches!(stopped, Error::SlashingProtectionUnusable { .. }), + "got {stopped}" + ); + assert!(shutdown.is_cancelled(), "the keymanager was told to stop"); + assert!( + finished.load(std::sync::atomic::Ordering::Acquire), + "its request in flight finishes before the client stops" + ); + } + + /// Duties that end on their own leave the same way: with the error that + /// ended them, after the keymanager has finished. + #[tokio::test(start_paused = true)] + async fn duties_that_stop_on_their_own_also_wait_for_the_keymanager() { + let protection = slashing_protection::SlashingProtection::in_memory(); + let shutdown = CancellationToken::new(); + let (keymanager, finished) = + keymanager_finishing_in(&shutdown, std::time::Duration::from_secs(5)); + + let stopped = run_until_stopped( + async { Error::BeaconNodeSyncing }, + &protection, + shutdown, + Some(keymanager), + ) + .await; + assert!(matches!(stopped, Error::BeaconNodeSyncing), "got {stopped}"); + assert!(finished.load(std::sync::atomic::Ordering::Acquire)); + } + + /// A request that never ends does not keep a stopped client from being + /// restarted: the wait for the keymanager is bounded. + #[tokio::test(start_paused = true)] + async fn a_keymanager_that_never_finishes_does_not_hold_the_client_up() { + let protection = slashing_protection::SlashingProtection::in_memory(); + let keymanager = tokio::spawn(std::future::pending::<()>()); + + let before = tokio::time::Instant::now(); + let stopped = run_until_stopped( + async { Error::BeaconNodeSyncing }, + &protection, + CancellationToken::new(), + Some(keymanager), + ) + .await; + assert!(matches!(stopped, Error::BeaconNodeSyncing), "got {stopped}"); + assert_eq!(before.elapsed(), KEYMANAGER_DRAIN); + } + + /// With no keymanager there is nothing to wait for. + #[tokio::test(start_paused = true)] + async fn stopping_without_a_keymanager_waits_for_nothing() { + let protection = slashing_protection::SlashingProtection::in_memory(); + let before = tokio::time::Instant::now(); + let stopped = run_until_stopped( + async { Error::BeaconNodeSyncing }, + &protection, + CancellationToken::new(), + None, + ) + .await; + assert!(matches!(stopped, Error::BeaconNodeSyncing), "got {stopped}"); + assert_eq!(before.elapsed(), std::time::Duration::ZERO); + } } diff --git a/crates/validator/src/proposal.rs b/crates/validator/src/proposal.rs index c16319f07..0c7258018 100644 --- a/crates/validator/src/proposal.rs +++ b/crates/validator/src/proposal.rs @@ -680,7 +680,7 @@ mod tests { let protection = || { let opened = SlashingProtection::open(dir.path(), Root::ZERO, &[pubkey], true).expect("opens"); - Arc::new(std::sync::Mutex::new(opened)) + crate::slashing_protection::SharedProtection::new(opened) }; let node = Arc::new(MockBeaconNode::new().with_block(slot(), 7)); @@ -716,7 +716,7 @@ mod tests { node.clone(), context(), settings, - Arc::new(std::sync::Mutex::new(opened)), + crate::slashing_protection::SharedProtection::new(opened), ) .propose(slot(), &duty(&pubkey, 7, slot()), &store) .await diff --git a/crates/validator/src/slashing_protection/mod.rs b/crates/validator/src/slashing_protection/mod.rs index d462d490e..15f8641b0 100644 --- a/crates/validator/src/slashing_protection/mod.rs +++ b/crates/validator/src/slashing_protection/mod.rs @@ -65,6 +65,7 @@ use std::sync::{Arc, Mutex}; use ethlambda_types::beacon::containers::shared::AttestationData; use ethlambda_types::beacon::primitives::{BlsPubkey, Epoch, Root, Slot}; +use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use crate::beacon_node::dto::encode_hex; @@ -76,7 +77,40 @@ use crate::error::{Error, Result}; /// a change, a file write, with no `.await` inside. Async callers go through /// [`blocking`], so neither the write nor waiting for another holder's write /// holds up the executor. -pub type SharedProtection = Arc>; +/// +/// It also carries the signal that the record has become unusable. A panic +/// while the lock is held poisons it for the life of the process, since +/// nothing here recovers a record a panic may have left half-updated in +/// memory: from then on no attestation and no block can be signed. The +/// remedy is a restart, which reads the record back from disk, where every +/// write either replaced it whole or left it alone. [`blocking`] raises the +/// signal from the access that poisons the lock, or first finds it poisoned, +/// and [`crate::run`] stops the client on it, for its supervisor to restart. +#[derive(Clone)] +pub struct SharedProtection { + record: Arc>, + unusable: CancellationToken, +} + +impl SharedProtection { + pub fn new(protection: SlashingProtection) -> Self { + Self { + record: Arc::new(Mutex::new(protection)), + unusable: CancellationToken::new(), + } + } + + /// Lock the record on the calling thread, waiting out any other holder. + /// For synchronous callers; an async one goes through [`blocking`]. + pub fn lock(&self) -> std::sync::LockResult> { + self.record.lock() + } + + /// Resolves once the record is unusable, and never on a healthy one. + pub async fn unusable(&self) { + self.unusable.cancelled().await; + } +} /// How long [`blocking`] waits for the record before refusing, and how often /// it looks. A flush takes milliseconds on a healthy disk; a quarter of a @@ -94,13 +128,12 @@ const RECORD_POLL: std::time::Duration = std::time::Duration::from_millis(5); /// The operation's own error comes back as it is, so a caller can still tell /// a write that failed (`Error::Io`, a disk to look at) from a record that /// could not be reached at all (`Error::SlashingProtectionUnusable`): the -/// lock is poisoned, because an operation panicked while holding it, and -/// stays so until the process restarts, since nothing here recovers a record -/// that a panic may have left half-updated in memory. The call whose own -/// operation panicked sees it as the blocking task's `JoinError` and is told -/// so, with the panic's message; every later call is told an earlier access -/// panicked. The duty loop asks [`ensure_usable`] once a slot and stops the -/// client on a poisoned record, so the restart comes from its supervisor. +/// lock is poisoned, because an operation panicked while holding it. The call +/// whose own operation panicked sees that as the blocking task's `JoinError` +/// and is told the panic was its own; every later call is told an earlier +/// access panicked. Either way the access raises the handle's signal (see +/// [`SharedProtection`]), from the blocking task rather than from the caller, +/// who may be gone by then. /// /// The operation runs to completion even if the caller is dropped, as a duty /// abandoned at its deadline is: that is safe, since a record written for a @@ -115,8 +148,12 @@ pub async fn blocking( protection: &SharedProtection, operation: impl FnOnce(&mut SlashingProtection) -> Result + Send + 'static, ) -> Result { - let protection = protection.clone(); + let SharedProtection { record, unusable } = protection.clone(); tokio::task::spawn_blocking(move || { + // Declared before the guard, so on a panic in `operation` it drops + // after the guard has poisoned the lock: nobody is told the record is + // unusable before it is. + let _signal = SignalIfPanicking(&unusable); // A bounded wait rather than `lock()`: a write normally holds the // record for milliseconds, so a wait this long means a flush is // stuck, and refusing then keeps a stuck disk from piling up one @@ -124,9 +161,12 @@ pub async fn blocking( // on a refusal. let deadline = std::time::Instant::now() + RECORD_WAIT; let mut protection = loop { - match protection.try_lock() { + match record.try_lock() { Ok(guard) => break guard, - Err(std::sync::TryLockError::Poisoned(_)) => return Err(poisoned_earlier()), + Err(std::sync::TryLockError::Poisoned(_)) => { + unusable.cancel(); + return Err(poisoned_earlier()); + } Err(std::sync::TryLockError::WouldBlock) => { if std::time::Instant::now() >= deadline { return Err(Error::SlashingProtectionUnusable { @@ -145,58 +185,43 @@ pub async fn blocking( }) .await .map_err(|err| { - let reason = match err.try_into_panic() { + let reason = if err.is_panic() { // This call's own operation panicked, and it held the record when // it did: nothing else in the task can panic. Said as this - // access's failure, with the panic's message, since "an earlier - // access" would send whoever reads it looking for another event. - Ok(panic) => format!( - "this access panicked while holding the record ({}), so nothing is signed until \ - the validator client is restarted", - panic_message(panic.as_ref()) - ), - Err(err) => format!("the record could not be consulted: {err}"), + // access's failure, since "an earlier access" would send whoever + // reads it looking for another event. The panic's message and + // location are what the panic hook printed, so they are not + // repeated here. + "this access panicked while holding the record, so no attestation or block is signed \ + until the validator client is restarted; the panic is reported just before this" + .to_string() + } else { + format!("the record could not be consulted: {err}") }; Error::SlashingProtectionUnusable { reason } })? } +/// Raises the record's signal when dropped by a panic unwinding past it. +struct SignalIfPanicking<'a>(&'a CancellationToken); + +impl Drop for SignalIfPanicking<'_> { + fn drop(&mut self) { + if std::thread::panicking() { + self.0.cancel(); + } + } +} + /// What a caller is told when it finds the record's lock already poisoned. fn poisoned_earlier() -> Error { Error::SlashingProtectionUnusable { - reason: "an earlier access panicked while holding the record, so nothing is signed \ - until the validator client is restarted" + reason: "an earlier access panicked while holding the record, so no attestation or block \ + is signed until the validator client is restarted" .to_string(), } } -/// A panic's message, when it carried one: `panic!` with a literal gives a -/// `&str` and with arguments a `String`. Anything else was raised with -/// `panic_any` and has no text to show. -fn panic_message(panic: &(dyn std::any::Any + Send)) -> &str { - panic - .downcast_ref::<&'static str>() - .copied() - .or_else(|| panic.downcast_ref::().map(String::as_str)) - .unwrap_or("no message") -} - -/// Refuse to go on with a record whose lock is poisoned. -/// -/// A poisoned record refuses every duty and never recovers (see -/// [`blocking`]), so a client left running on one looks alive to whatever -/// supervises it while signing nothing. The duty loop asks this once a slot -/// and stops on an `Err`, which ends the process with a failure status: -/// systemd or Kubernetes then restarts it, and the restart reads the record -/// back from disk, which every write either replaced whole or left alone. -/// That is the remedy the refusals ask for. -pub fn ensure_usable(protection: &SharedProtection) -> Result<()> { - if protection.is_poisoned() { - return Err(poisoned_earlier()); - } - Ok(()) -} - /// Whether an attestation's source epoch is at or past its target, which is /// never valid outside genesis: with unsigned epochs, `(0, 0)` is the only /// such pair that is. One predicate for the three places that refuse it, @@ -568,11 +593,11 @@ impl SlashingProtection { /// production caller can reach it. #[cfg(any(test, feature = "test-utils"))] pub fn in_memory() -> SharedProtection { - Arc::new(Mutex::new(Self { + SharedProtection::new(Self { genesis_validators_root: Root::ZERO, records: HashMap::new(), storage: None, - })) + }) } /// The record's path, or `None` for the in-memory record of the tests. @@ -913,11 +938,13 @@ mod tests { .expect("the record is written") } + /// What [`SlashingProtection::in_memory`] shares, held directly. fn memory() -> SlashingProtection { - Arc::into_inner(SlashingProtection::in_memory()) - .expect("sole owner") - .into_inner() - .expect("not poisoned") + SlashingProtection { + genesis_validators_root: Root::ZERO, + records: HashMap::new(), + storage: None, + } } fn open(dir: &Path, keys: &[u8], init: bool) -> Result { @@ -1584,14 +1611,14 @@ mod tests { // A poisoned record. - /// The duty loop's once-a-slot question: a healthy record lets it go on, - /// a poisoned one stops the client, and the error is not one to retry. - #[test] - fn a_poisoned_record_stops_the_client() { - let protection = SlashingProtection::in_memory(); - ensure_usable(&protection).expect("a healthy record is usable"); + /// How long a test waits for a signal that should have been raised. + const RAISED: std::time::Duration = std::time::Duration::from_secs(5); + /// How long a test watches for a signal that should not be. + const QUIET: std::time::Duration = std::time::Duration::from_millis(20); - // Poisoned the way a bug would: a panic while the lock is held. + /// Poisoned the way a bug would poison it: a panic while the lock is + /// held. Around [`blocking`], so nothing has observed it yet. + fn poison(protection: &SharedProtection) { let held = protection.clone(); std::thread::spawn(move || { let _guard = held.lock().expect("not poisoned yet"); @@ -1599,49 +1626,82 @@ mod tests { }) .join() .expect_err("the thread panics"); - - let err = ensure_usable(&protection).expect_err("a poisoned record is not"); - assert!( - matches!(err, Error::SlashingProtectionUnusable { .. }), - "got {err}" - ); - assert!(!err.is_retryable(), "the loop must stop, not try again"); } - /// The access whose own operation panics is told it was its panic, with - /// the message, and only the ones after it that an earlier access - /// panicked: the same wording for both points the first at an event that - /// never happened. + /// The access whose own operation panics is told the panic was its own, + /// and only the ones after it that an earlier access panicked: the same + /// wording for both points the first at an event that never happened. It + /// also raises the signal the client stops on, with nothing else having + /// to touch the record first. #[tokio::test] - async fn the_access_that_panics_is_told_so_and_later_ones_that_an_earlier_one_did() { + async fn the_access_that_panics_is_told_so_and_raises_the_signal() { let protection = SlashingProtection::in_memory(); + blocking(&protection, |_| Ok(())) + .await + .expect("a healthy record answers"); + tokio::time::timeout(QUIET, protection.unusable()) + .await + .expect_err("a healthy record raises nothing"); let own = blocking(&protection, |_| -> Result<()> { panic!("a bug in the record") }) .await - .expect_err("the operation panicked") - .to_string(); - assert!(own.contains("this access panicked"), "{own}"); - assert!(own.contains("a bug in the record"), "{own}"); + .expect_err("the operation panicked"); + assert!( + matches!(own, Error::SlashingProtectionUnusable { .. }), + "got {own}" + ); + assert!(!own.is_retryable(), "nothing to try again"); + assert!(own.to_string().contains("this access panicked"), "{own}"); + tokio::time::timeout(RAISED, protection.unusable()) + .await + .expect("the panicking access raises the signal"); let later = blocking(&protection, |_| Ok(())) .await .expect_err("the record is poisoned") .to_string(); assert!(later.contains("an earlier access panicked"), "{later}"); - ensure_usable(&protection).expect_err("and the duty loop stops on it"); } - /// A panic raised with arguments carries a `String`, not a `&str`, and - /// one raised with `panic_any` carries neither. - #[test] - fn a_panics_message_is_read_whatever_it_was_raised_with() { - let literal: Box = Box::new("a literal"); - assert_eq!(panic_message(literal.as_ref()), "a literal"); - let formatted: Box = Box::new(format!("slot {}", 7)); - assert_eq!(panic_message(formatted.as_ref()), "slot 7"); - let neither: Box = Box::new(7_u64); - assert_eq!(panic_message(neither.as_ref()), "no message"); + /// A lock poisoned without anyone seeing it raises nothing until an + /// access finds it so, and that access raises the signal. + #[tokio::test] + async fn the_access_that_finds_the_record_poisoned_raises_the_signal() { + let protection = SlashingProtection::in_memory(); + poison(&protection); + tokio::time::timeout(QUIET, protection.unusable()) + .await + .expect_err("nothing has looked at the record yet"); + + blocking(&protection, |_| Ok(())) + .await + .expect_err("the record is poisoned"); + tokio::time::timeout(RAISED, protection.unusable()) + .await + .expect("the access that found it raises the signal"); + } + + /// The signal comes from the blocking task, not from the caller awaiting + /// it: a duty abandoned at its deadline drops the caller and leaves the + /// task running, and a panic after that must still stop the client. + #[tokio::test] + async fn the_signal_is_raised_even_when_the_caller_is_gone() { + let protection = SlashingProtection::in_memory(); + let (release, held) = std::sync::mpsc::channel::<()>(); + let call = blocking(&protection, move |_| -> Result<()> { + held.recv().expect("released"); + panic!("a bug after the caller left") + }); + // Polled, so the task is running, then dropped at the timeout. + tokio::time::timeout(QUIET, call) + .await + .expect_err("the operation is still held"); + + release.send(()).expect("the task is still running"); + tokio::time::timeout(RAISED, protection.unusable()) + .await + .expect("the task itself raises the signal"); } } diff --git a/docs/cli.md b/docs/cli.md index 172d96c8b..64c4ac36e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -280,9 +280,10 @@ implementation's, so none of the [common flags](#common-flags) apply to it. > is an `flock`, which a network filesystem may not honour. Unix only. Run the > client under a supervisor that restarts it (systemd's `Restart=on-failure`, > a Kubernetes pod): if a bug ever panics while the record is held, the record -> is unusable for the rest of that process, every duty is refused, and the -> client exits with a failure at the next slot instead of idling, so that the -> restart rereads the record from disk. A key with no record is refused at startup unless +> is unusable for the rest of that process and no attestation or block can be +> signed, so the client stops at once and exits with a failure, after letting +> a keymanager request in flight finish, and the restart rereads the record +> from disk. A key with no record is refused at startup unless > `--init-slashing-protection` is passed. A key moving here from another client > brings its history with [`slashing-protection import`](#slashing-protection), > or in the same keymanager API request that imports the key. See From 346b52399278e050079d90cdbd514bb7d8299ade Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Thu, 12 Nov 2026 17:31:15 -0300 Subject: [PATCH 5/6] Say what a failed import's undo left on disk When registering an imported key failed and the undo failed too, the answer and the log said the next start would refuse the key, whichever step of the undo had failed. That is true only when the definitions entry is still there. When the entry was removed and only the files stayed, the next start is fine and the files are merely left behind, unused. undo_import now says which step failed: UndoFailure::Definitions, where nothing was undone, or UndoFailure::Files, naming the keys whose files could not be put back. Each key's status is built from that (unregistered_status) and says what is on disk for that key, and the log line warns of a refused start only where one follows. Two changes on the way. The undo no longer stops at the first key whose files cannot be put back, so the others' files still go back. And an Unflushed definitions save inside the undo is no longer returned as a failure before the files are touched: the file was already replaced, so it warns and goes on. --- crates/validator/src/http_api/keystores.rs | 332 ++++++++++++++++++--- 1 file changed, 290 insertions(+), 42 deletions(-) diff --git a/crates/validator/src/http_api/keystores.rs b/crates/validator/src/http_api/keystores.rs index 1dce5dd7a..43bf11cf0 100644 --- a/crates/validator/src/http_api/keystores.rs +++ b/crates/validator/src/http_api/keystores.rs @@ -388,12 +388,18 @@ pub async fn import( undo_import(&context, &mut definitions, &added, &replaced, &files) } }; - if let Err(err) = &rolled_back { - tracing::error!( + match &rolled_back { + Ok(()) => {} + Err(UndoFailure::Definitions(err)) => tracing::error!( %err, - "Could not undo a failed import; the next start refuses its keys until they have a \ - slashing-protection record" - ); + "Could not undo a failed import; the next start refuses the keys it added until \ + they have a slashing-protection record" + ), + Err(UndoFailure::Files(left)) => tracing::error!( + keys = left.len(), + "A failed import's definitions entries are undone, so the next start loads nothing \ + it added, but not all of its keystore and password files could be put back" + ), } // The definitions guard stays held across the store lock, in the order @@ -406,34 +412,11 @@ pub async fn import( .into_iter() .map(|result| match (result, ®istration) { (Ok((_, secret, _)), Ok(_)) => finish_import(&mut store, &secret), - (Ok((pubkey, _, _)), Err(err)) => { - match (before.contains_key(&pubkey), rolled_back.is_ok()) { - // A key the file already listed, undone: its keystore, - // its password and its entry are what they were before - // the request, and nothing about it changed in memory, so - // it stays whatever it was, loaded since startup in the - // ordinary case. - (true, true) => KeyStatus::error(format!( - "not imported: its slashing-protection record could not be written: \ - {err}; its keystore and definitions entry are back as they were" - )), - // The undo failed as well, so what this request wrote - // over the key's files and entry may still be there. - (true, false) => KeyStatus::error(format!( - "keystore rewritten, but the slashing-protection record could not be \ - written: {err}; its definitions entry could not be put back" - )), - (false, true) => KeyStatus::error(format!( - "not imported: its slashing-protection record could not be written: {err}" - )), - // Still on disk, not loaded now, and refused at the next - // start until it has a record. - (false, false) => KeyStatus::error(format!( - "saved, but not loaded: its slashing-protection record could not be \ - written: {err}" - )), - } - } + (Ok((pubkey, _, _)), Err(err)) => unregistered_status( + before.contains_key(&pubkey), + Undone::of(&rolled_back, &pubkey), + err, + ), (Err(status), _) => status, }) .collect(); @@ -443,6 +426,83 @@ pub async fn import( Json(StatusResponse { data }) } +/// What [`undo_import`] could not undo, by the step that failed, since the +/// two leave different things on disk. +#[derive(Debug)] +enum UndoFailure { + /// The definitions file could not be saved, so nothing was undone: the + /// entries the request wrote are the ones on disk, and so are its + /// keystore and password files. A key it added is listed with no record, + /// which the next start refuses. + Definitions(Error), + /// The definitions file is back as it was, so the next start loads + /// nothing the request added, but the keystore or password file of these + /// keys could not be put back. + Files(Vec<(BlsPubkey, Error)>), +} + +/// What became of one saved key when its registration failed and the import +/// was undone. +#[derive(Debug, Clone, Copy)] +enum Undone<'a> { + /// Its entry and its files are what they were before the request. + Fully, + /// Nothing was put back: its entry and its files are the request's. + NotAtAll, + /// Its entry is what it was before the request; its files are not, for + /// this reason. + EntryOnly(&'a Error), +} + +impl<'a> Undone<'a> { + fn of(rolled_back: &'a std::result::Result<(), UndoFailure>, pubkey: &BlsPubkey) -> Self { + match rolled_back { + Ok(()) => Self::Fully, + Err(UndoFailure::Definitions(_)) => Self::NotAtAll, + Err(UndoFailure::Files(left)) => left + .iter() + .find(|(key, _)| key == pubkey) + .map_or(Self::Fully, |(_, err)| Self::EntryOnly(err)), + } + } +} + +/// The status of a key that was saved but could not be registered: what is on +/// disk for it now, which depends on how far the undo got. `listed_before` is +/// whether the definitions file already listed the key, in which case the +/// request rewrote its keystore rather than added one, and nothing about the +/// key changed in memory: it stays whatever it was, loaded since startup in +/// the ordinary case. +fn unregistered_status(listed_before: bool, undone: Undone<'_>, err: &Error) -> KeyStatus { + let cause = format!("its slashing-protection record could not be written: {err}"); + KeyStatus::error(match (listed_before, undone) { + (true, Undone::Fully) => format!( + "not imported: {cause}; its keystore and definitions entry are back as they were" + ), + (false, Undone::Fully) => format!("not imported: {cause}"), + (true, Undone::NotAtAll) => format!( + "keystore rewritten, but {cause}; the rewrite could not be undone, so its \ + definitions entry, keystore and password are the ones this request wrote" + ), + // Still listed and enabled on disk, with no record: not loaded now, + // and `SlashingProtection::open` refuses the whole client for it. + (false, Undone::NotAtAll) => format!( + "saved, but not loaded: {cause}; it could not be taken off disk again, so the \ + client will not start again until the key has a record or is removed" + ), + (true, Undone::EntryOnly(undo)) => format!( + "not imported: {cause}; its definitions entry is back as it was, but its keystore \ + and password files could not be put back ({undo}) and may be the ones this \ + request wrote" + ), + // Orphans: nothing lists them, so nothing loads them. + (false, Undone::EntryOnly(undo)) => format!( + "not imported: {cause}; its definitions entry is removed, but its keystore and \ + password files could not be ({undo}) and are left behind, unused" + ), + }) +} + /// Undo what a failed import did: the definitions entries it added are /// removed and the ones it replaced put back, in one save, then every /// keystore and password file it wrote goes back to what was at its path @@ -450,13 +510,16 @@ pub async fn import( /// first, since it is what the next start loads from, so a crash in between /// leaves orphan files and never a key that would be loaded without a /// record. A file that is already gone is not an error. +/// +/// Reports which step failed. A key whose files cannot be put back does not +/// stop the others': each is tried, and the ones left are named. fn undo_import( context: &KeymanagerContext, definitions: &mut ValidatorDefinitions, added: &[BlsPubkey], replaced: &[(BlsPubkey, ValidatorDefinition)], files: &[(BlsPubkey, &Overwritten)], -) -> crate::Result<()> { +) -> std::result::Result<(), UndoFailure> { if added.is_empty() && replaced.is_empty() { return Ok(()); } @@ -472,19 +535,42 @@ fn undo_import( *definition = previous.clone(); } } - save_definitions(context, definitions)?; + match save_definitions(context, definitions) { + Ok(()) => {} + // The file is replaced and every reader sees it, so the entries are + // undone and the files follow. Only its survival of a power loss is + // in doubt, and the file that would come back lists keys with no + // record, which the next start refuses. + Err(Error::Unflushed { path, source }) => warn!( + %path, + %source, + "Validator definitions put back, but the directory could not be flushed" + ), + Err(err) => return Err(UndoFailure::Definitions(err)), + } // Newest first: a key listed twice in one request snapshotted, the second // time, what the first time wrote, so the earliest snapshot must be the // one that lands last. + let mut left = Vec::new(); for (pubkey, overwritten) in files.iter().rev() { - let [keystore, password] = written_paths(context, pubkey)?; - put_back(&keystore, overwritten.keystore.as_deref())?; - put_back( - &password, - overwritten.password.as_deref().map(|bytes| &bytes[..]), - )?; + // The password only after the keystore: one that stays the request's + // keeps the password it was encrypted under. + let put = written_paths(context, pubkey).and_then(|[keystore, password]| { + put_back(&keystore, overwritten.keystore.as_deref())?; + put_back( + &password, + overwritten.password.as_deref().map(|bytes| &bytes[..]), + ) + }); + if let Err(err) = put { + left.push((*pubkey, err)); + } + } + if left.is_empty() { + Ok(()) + } else { + Err(UndoFailure::Files(left)) } - Ok(()) } /// Put `path` back to `previous`: its former bytes, or nothing. @@ -1638,6 +1724,168 @@ mod tests { assert_eq!(definitions.0, vec![previous]); } + /// The entry an import writes for `pubkey` in `dir`. + fn imported_entry(dir: &std::path::Path, pubkey: &BlsPubkey) -> ValidatorDefinition { + let base = encode_hex(&pubkey.0); + ValidatorDefinition { + enabled: true, + voting_public_key: base.clone(), + voting_keystore_path: dir.join(format!("{base}.json")), + voting_keystore_password_path: dir.join(base), + } + } + + /// A non-empty directory at `path`: neither a file to remove nor a place + /// a rename can land, whoever runs the test. + fn obstruct(path: &std::path::Path) { + std::fs::create_dir(path).expect("a directory in its place"); + std::fs::write(path.join("inside"), b"x").expect("writes"); + } + + /// The undo's two steps leave different things on disk, and it says which + /// one failed. Here the definitions are saved and one key's keystore + /// cannot be removed: both entries are undone, that key is named, and the + /// other key's files still go, since one failure does not stop the rest. + #[test] + fn an_undo_names_the_keys_whose_files_it_could_not_put_back() { + let dir = tempfile::tempdir().expect("temp dir"); + let context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); + let (stuck, fine) = (BlsPubkey([1; 48]), BlsPubkey([2; 48])); + let stuck_entry = imported_entry(dir.path(), &stuck); + let fine_entry = imported_entry(dir.path(), &fine); + obstruct(&stuck_entry.voting_keystore_path); + std::fs::write(&stuck_entry.voting_keystore_password_path, b"password").expect("writes"); + std::fs::write(&fine_entry.voting_keystore_path, b"keystore").expect("writes"); + std::fs::write(&fine_entry.voting_keystore_password_path, b"password").expect("writes"); + let mut definitions = ValidatorDefinitions(vec![stuck_entry.clone(), fine_entry.clone()]); + definitions.save(dir.path()).expect("saves"); + + // Neither key had files before. Undone newest first, so `stuck` + // fails before `fine` is reached. + let nothing = Overwritten { + keystore: None, + password: None, + }; + let files = [(fine, ¬hing), (stuck, ¬hing)]; + let failure = undo_import(&context, &mut definitions, &[stuck, fine], &[], &files) + .expect_err("one keystore cannot be removed"); + + let UndoFailure::Files(left) = &failure else { + panic!("the definitions were saved, got {failure:?}"); + }; + assert_eq!( + left.iter().map(|(key, _)| *key).collect::>(), + vec![stuck] + ); + assert!( + ValidatorDefinitions::open(dir.path()) + .expect("opens") + .0 + .is_empty(), + "both entries are undone, so the next start loads neither key" + ); + assert!( + !fine_entry.voting_keystore_path.exists() + && !fine_entry.voting_keystore_password_path.exists(), + "the other key's files still go" + ); + assert!( + stuck_entry.voting_keystore_password_path.exists(), + "a keystore that stays keeps the password it was encrypted under" + ); + + let rolled_back = Err(failure); + assert!(matches!( + Undone::of(&rolled_back, &stuck), + Undone::EntryOnly(_) + )); + assert!(matches!(Undone::of(&rolled_back, &fine), Undone::Fully)); + } + + /// The other step: the definitions file cannot be saved, so nothing is + /// undone, the files the entry names are left where they are, and every + /// key of the request is reported that way. + #[test] + fn an_undo_that_cannot_save_the_definitions_undoes_nothing() { + let dir = tempfile::tempdir().expect("temp dir"); + let context = context_with_dirs(Arc::new(RwLock::new(ValidatorStore::new())), dir.path()); + let key = BlsPubkey([1; 48]); + let entry = imported_entry(dir.path(), &key); + std::fs::write(&entry.voting_keystore_path, b"keystore").expect("writes"); + std::fs::write(&entry.voting_keystore_password_path, b"password").expect("writes"); + let mut definitions = ValidatorDefinitions(vec![entry.clone()]); + obstruct(&dir.path().join(crate::keys::definitions::DEFINITIONS_FILE)); + + let nothing = Overwritten { + keystore: None, + password: None, + }; + let failure = undo_import(&context, &mut definitions, &[key], &[], &[(key, ¬hing)]) + .expect_err("the definitions cannot be saved"); + + assert!( + matches!(failure, UndoFailure::Definitions(_)), + "got {failure:?}" + ); + assert!( + entry.voting_keystore_path.exists() && entry.voting_keystore_password_path.exists(), + "the files a still-listed entry names are not removed from under it" + ); + assert!(matches!(Undone::of(&Err(failure), &key), Undone::NotAtAll)); + } + + /// Each way the undo can end gets a status that says what is on disk for + /// the key, and why the files stayed when they did. + #[test] + fn the_status_of_an_unregistered_key_says_what_is_on_disk() { + let err = Error::SlashingProtectionUnusable { + reason: "the record's reason".to_string(), + }; + let undo = Error::Keystore { + path: "a path".to_string(), + reason: "the undo's reason".to_string(), + }; + let message = |listed_before, undone| { + let status = unregistered_status(listed_before, undone, &err); + assert_eq!(status.status, "error"); + let message = status.message.expect("an error says why"); + assert!(message.contains("the record's reason"), "{message}"); + message + }; + + let added = message(false, Undone::Fully); + assert!(added.starts_with("not imported"), "{added}"); + let replaced = message(true, Undone::Fully); + assert!( + replaced.ends_with("its keystore and definitions entry are back as they were"), + "{replaced}" + ); + + let added = message(false, Undone::NotAtAll); + assert!(added.starts_with("saved, but not loaded"), "{added}"); + assert!(added.contains("will not start again"), "{added}"); + let replaced = message(true, Undone::NotAtAll); + assert!(replaced.starts_with("keystore rewritten"), "{replaced}"); + assert!(replaced.contains("could not be undone"), "{replaced}"); + + let added = message(false, Undone::EntryOnly(&undo)); + assert!(added.starts_with("not imported"), "{added}"); + assert!( + added.contains("its definitions entry is removed"), + "{added}" + ); + assert!(added.contains("left behind"), "{added}"); + assert!(added.contains("the undo's reason"), "{added}"); + let replaced = message(true, Undone::EntryOnly(&undo)); + assert!(replaced.starts_with("not imported"), "{replaced}"); + assert!( + replaced.contains("its definitions entry is back as it was"), + "{replaced}" + ); + assert!(replaced.contains("could not be put back"), "{replaced}"); + assert!(replaced.contains("the undo's reason"), "{replaced}"); + } + /// A request whose history does not parse as an interchange is refused /// whole, importing nothing: `{}` has neither metadata nor data. History /// that does parse is imported with its keys (see From 36277cb443ff78c40a778d6c18c9cf918f07ff0b Mon Sep 17 00:00:00 2001 From: Pablo Deymonnaz Date: Thu, 12 Nov 2026 17:31:26 -0300 Subject: [PATCH 6/6] Tighten the tests and the smaller points from the review - The interchange vector runner skips a refused case only when the importer's own error says the refusal is for an attestation whose source is at or after its target, on top of the step being flagged slashable and the interchange holding one. A refusal of such an interchange for any other reason now fails the case. - MockBeaconNode::without_attestation_data_check hands attestation data back unchecked, and the three service tests (wrong slot, wrong target epoch, invalid index) use it, so what refuses is attest's own re-check rather than the mock's. Deleting the re-check now fails all three. - validate_attestation_data's lean arm is an unreachable! naming the function, as the lean arms elsewhere are, instead of an error blamed on the beacon node. highest_attestation_index returns a plain u64 again. - validate_attestation_data's doc states that fork must be the fork of slot, and the failover tests derive it from mainnet's schedule instead of naming one. - an_export_needs_no_writable_directory is skipped under root, where a mode 0500 directory refuses nothing. - The vector runner reads epochs written as bare integers as well as strings. --- crates/validator/src/attestation.rs | 19 +++++-- crates/validator/src/beacon_node/fallback.rs | 17 +++++-- crates/validator/src/beacon_node/mock.rs | 19 ++++++- crates/validator/src/beacon_node/mod.rs | 42 ++++++++++------ .../validator/src/slashing_protection/mod.rs | 4 ++ crates/validator/tests/interchange_vectors.rs | 49 ++++++++++++------- 6 files changed, 107 insertions(+), 43 deletions(-) diff --git a/crates/validator/src/attestation.rs b/crates/validator/src/attestation.rs index 8e773610f..a857db8b2 100644 --- a/crates/validator/src/attestation.rs +++ b/crates/validator/src/attestation.rs @@ -705,7 +705,14 @@ mod tests { let store = RwLock::new(store); let pubkey_hex = encode_hex(&pubkey.0); - let node = Arc::new(MockBeaconNode::new().with_attestation_data(96)); + // A node that breaks the trait's contract and hands the stale data + // back unchecked, so the refusal below is `attest`'s own, the last + // one before a signature, and not the implementation's. + let node = Arc::new( + MockBeaconNode::new() + .with_attestation_data(96) + .without_attestation_data_check(), + ); let service = AttestationService::new(node.clone(), context(), SlashingProtection::in_memory()); @@ -734,7 +741,8 @@ mod tests { let store = RwLock::new(store); let pubkey_hex = encode_hex(&pubkey.0); - let mut node = MockBeaconNode::new(); + // Unchecked by the node, so `attest`'s own check is what refuses. + let mut node = MockBeaconNode::new().without_attestation_data_check(); node.attestation_data = Some(AttestationData { slot: 96, index: 0, @@ -769,6 +777,10 @@ mod tests { /// every peer would reject, and a correct answer for the same slot is /// still signed. `context()` is mainnet's schedule, where slot 96 is long /// before gloas, so 1 is not an index a vote there can carry. + /// + /// The node hands its answer back unchecked, so what refuses it is + /// `attest`'s own check, under the fork `attest` derives for the slot: + /// with that check gone, or run under gloas, the first call would sign. #[tokio::test] async fn an_index_invalid_at_the_slots_fork_is_refused_and_records_nothing() { let mut store = ValidatorStore::new(); @@ -780,7 +792,8 @@ mod tests { let wrong = Arc::new( MockBeaconNode::new() .with_attestation_data(96) - .with_attestation_index(1), + .with_attestation_index(1) + .without_attestation_data_check(), ); let err = AttestationService::new(wrong.clone(), context(), protection.clone()) .attest(96, &duties, &store) diff --git a/crates/validator/src/beacon_node/fallback.rs b/crates/validator/src/beacon_node/fallback.rs index 918386c50..3b8972009 100644 --- a/crates/validator/src/beacon_node/fallback.rs +++ b/crates/validator/src/beacon_node/fallback.rs @@ -436,6 +436,14 @@ mod tests { use crate::beacon_node::mock::MockBeaconNode; use ethlambda_types::beacon::primitives::Root; + /// The fork a caller asks about `slot` under: the one mainnet's schedule + /// puts it in, as `AttestationService` derives it, rather than one made + /// up here that the slot does not belong to. + fn fork_of(slot: Slot) -> ForkName { + let epoch = ethlambda_types::beacon::signing::compute_epoch_at_slot(slot); + ethlambda_types::beacon::config::Config::mainnet().fork_at_epoch(epoch) + } + fn healthy() -> MockBeaconNode { let mut node = MockBeaconNode::new(); node.genesis = Some(Genesis { @@ -474,7 +482,7 @@ mod tests { let fallback = FallbackBeaconNode::new(vec![stale, fresh]); let data = fallback - .attestation_data(96, ForkName::Electra) + .attestation_data(96, fork_of(96)) .await .expect("the second node answers correctly"); @@ -494,7 +502,7 @@ mod tests { ]); let err = fallback - .attestation_data(96, ForkName::Electra) + .attestation_data(96, fork_of(96)) .await .expect_err("no node answered about the requested slot"); @@ -506,7 +514,8 @@ mod tests { /// answer being signed, recorded and then rejected by every peer. #[tokio::test] async fn a_node_answering_with_an_index_invalid_at_the_fork_is_failed_over_from() { - // Electra fixed the index at zero; node 1 answers 1 regardless. + // Slot 96 is before gloas, where the index is zero; node 1 answers 1 + // regardless. let wrong = MockBeaconNode::new() .with_attestation_data(96) .with_attestation_index(1); @@ -514,7 +523,7 @@ mod tests { let fallback = FallbackBeaconNode::new(vec![wrong, right]); let data = fallback - .attestation_data(96, ForkName::Electra) + .attestation_data(96, fork_of(96)) .await .expect("the second node answers correctly"); diff --git a/crates/validator/src/beacon_node/mock.rs b/crates/validator/src/beacon_node/mock.rs index b5d46c907..6b9dae220 100644 --- a/crates/validator/src/beacon_node/mock.rs +++ b/crates/validator/src/beacon_node/mock.rs @@ -64,6 +64,9 @@ pub struct MockBeaconNode { pub duties: Mutex>, pub proposers: Mutex>, pub attestation_data: Option, + /// Hand `attestation_data` back without checking it, as an implementation + /// that failed to honour the trait's contract would. + pub attestation_data_unchecked: bool, /// The slot and proposer the mock will produce a block for. `None` makes /// `produce_block` fail, the way an unset `attestation_data` does. pub produces_block: Option<(Slot, ValidatorIndex)>, @@ -310,6 +313,15 @@ impl MockBeaconNode { .clone() } + /// Break the contract on `BeaconNodeApi::attestation_data`: answer with + /// whatever data is set, unchecked. For tests of the caller's own check, + /// which a mock that honours the contract never lets a wrong answer + /// reach. + pub fn without_attestation_data_check(mut self) -> Self { + self.attestation_data_unchecked = true; + self + } + /// Give `attestation_data` answers a nonzero `index`, the way a gloas node /// signals a full payload. pub fn with_attestation_index(mut self, index: u64) -> Self { @@ -598,8 +610,11 @@ impl BeaconNodeApi for MockBeaconNode { // with data for another slot behaves like a node stuck on a stale // head: it errors, and failover moves past it. Without this the mock // would be a more permissive node than any real one, and tests built - // on it would not reflect what happens in production. - crate::beacon_node::validate_attestation_data(slot, fork, &data)?; + // on it would not reflect what happens in production. Unless a test + // asks for exactly that, to reach the caller's own check. + if !self.attestation_data_unchecked { + crate::beacon_node::validate_attestation_data(slot, fork, &data)?; + } Ok(data) } diff --git a/crates/validator/src/beacon_node/mod.rs b/crates/validator/src/beacon_node/mod.rs index 5399d7ab3..30a813fce 100644 --- a/crates/validator/src/beacon_node/mod.rs +++ b/crates/validator/src/beacon_node/mod.rs @@ -56,6 +56,13 @@ pub mod mock; /// it, and the `Err` is one failover moves past to a node that may answer /// properly. Lodestar's validator client refuses the same before signing /// (`validateAttestationDuty`). +/// +/// `fork` must be the fork `slot` falls in on the caller's schedule, and is +/// taken on trust: an implementation is handed it with the request and holds +/// no schedule to check it against, so the bound on `index` is only as right +/// as the fork it is given. [`crate::attestation::AttestationService`], the +/// caller that signs, derives it from its own `Config` for the slot it asks +/// about, and runs this check again under it. pub fn validate_attestation_data(slot: Slot, fork: ForkName, data: &AttestationData) -> Result<()> { if data.slot != slot { return Err(crate::error::Error::InconsistentResponse(format!( @@ -83,7 +90,7 @@ pub fn validate_attestation_data(slot: Slot, fork: ForkName, data: &AttestationD data.source.epoch, data.target.epoch ))); } - let highest_index = highest_attestation_index(fork)?; + let highest_index = highest_attestation_index(fork); if data.index > highest_index { return Err(crate::error::Error::InconsistentResponse(format!( "attestation data for slot {slot} carries index {}, and a {} attestation's is at \ @@ -100,7 +107,7 @@ pub fn validate_attestation_data(slot: Slot, fork: ForkName, data: &AttestationD /// Every fork is named rather than compared against, so a fork added after /// gloas has to say what its index means instead of inheriting gloas's /// answer. -fn highest_attestation_index(fork: ForkName) -> Result { +fn highest_attestation_index(fork: ForkName) -> u64 { match fork { // A committee, and the one the query names: this client only ever // asks about committee 0 (see `BeaconNodeApi::attestation_data`), so @@ -109,22 +116,25 @@ fn highest_attestation_index(fork: ForkName) -> Result { | ForkName::Altair | ForkName::Bellatrix | ForkName::Capella - | ForkName::Deneb => Ok(0), + | ForkName::Deneb => 0, // EIP-7549 moved the committee out of the signed data and left the // field fixed at zero. - ForkName::Electra | ForkName::Fulu => Ok(0), + ForkName::Electra | ForkName::Fulu => 0, // The payload signal: 0 for a same-slot block or an empty one, 1 for // a block whose payload was seen. Which of the two is right depends // on the node's view of the block, which this client does not have, // so only the range is checked. - ForkName::Gloas => Ok(1), + ForkName::Gloas => 1, // Not a Beacon Chain fork: `Config::fork_at_epoch` never names it, so - // a caller that does is wrong, and nothing is signed for it. - ForkName::Lean => Err(crate::error::Error::InconsistentResponse( - "attestation data was asked for under the lean fork, which is not a Beacon Chain \ - fork and has no attestation of this shape" - .to_string(), - )), + // a caller that does has a bug of its own. A panic rather than an + // `Err`, because this runs after the node has answered: an `Err` + // here is `InconsistentResponse`, which failover blames on the node + // and retries on every other one, so a local bug would read as every + // beacon node failing, every slot. + ForkName::Lean => unreachable!( + "validate_attestation_data: lean is not a Beacon Chain fork and has no attestation \ + data of this shape" + ), } } @@ -730,10 +740,12 @@ mod tests { assert!(validate_attestation_data(100, fork, &with_index(u64::MAX)).is_err()); } - /// Lean is not on the Beacon Chain's fork timeline, so no index is valid - /// for it and nothing is signed. + /// Lean is not on the Beacon Chain's fork timeline and no schedule names + /// it, so a caller that does has a bug, reported as one rather than as a + /// node answering badly. #[test] - fn the_lean_fork_has_no_valid_index() { - assert!(validate_attestation_data(100, ForkName::Lean, &with_index(0)).is_err()); + #[should_panic(expected = "lean is not a Beacon Chain fork")] + fn the_lean_fork_is_a_bug_in_the_caller() { + let _ = validate_attestation_data(100, ForkName::Lean, &with_index(0)); } } diff --git a/crates/validator/src/slashing_protection/mod.rs b/crates/validator/src/slashing_protection/mod.rs index 15f8641b0..a76f7ea0e 100644 --- a/crates/validator/src/slashing_protection/mod.rs +++ b/crates/validator/src/slashing_protection/mod.rs @@ -1533,6 +1533,10 @@ mod tests { fn an_export_needs_no_writable_directory() { use std::os::unix::fs::PermissionsExt as _; + if running_as_root() { + eprintln!("skipped: root bypasses directory permissions"); + return; + } let dir = tempfile::tempdir().expect("temp dir"); let mut protection = open(dir.path(), &[1], true).expect("init"); propose(&mut protection, 1, 97).expect("propose"); diff --git a/crates/validator/tests/interchange_vectors.rs b/crates/validator/tests/interchange_vectors.rs index 4823dd51d..19d6e11c9 100644 --- a/crates/validator/tests/interchange_vectors.rs +++ b/crates/validator/tests/interchange_vectors.rs @@ -122,34 +122,36 @@ fn run(case: &Case) -> Result<(), String> { SlashingProtection::open(dir.path(), root, &keys, true).map_err(|err| err.to_string())?; for (index, step) in case.steps.iter().enumerate() { - let imported = protection.import(&step.interchange.to_string()); - match (imported.is_ok(), step.should_succeed) { - (true, true) | (false, false) => {} + let imported = protection + .import(&step.interchange.to_string()) + .map_err(|err| err.to_string()); + match (&imported, step.should_succeed) { + (Ok(_), true) | (Err(_), false) => {} // The README lets an importer refuse an interchange holding // slashable data, in which case the rest of the case is skipped. // This one absorbs all of it but an attestation whose source is at - // or past its target, so only a step whose interchange holds one - // may be refused, which ends the case, as it does in Lighthouse. - // Every other slashable step must import: skipping on the flag - // alone would pass a suite whose importer had started refusing, + // or past its target, so a refusal ends the case, as it does in + // Lighthouse, only when the step's interchange holds one and that + // is what the importer says it refused. Every other refusal is a + // failure: skipping on the flag alone, or on the interchange + // alone, would pass a suite whose importer had started refusing, // say, a duplicated pubkey or a surround, with that case's blocks // and attestations never attempted. - (false, true) - if step.contains_slashable_data && holds_impossible_source(&step.interchange) => + (Err(refusal), true) + if step.contains_slashable_data + && holds_impossible_source(&step.interchange) + && refusal.contains(IMPOSSIBLE_SOURCE_REFUSAL) => { return Ok(()); } - (false, true) => { + (Err(refusal), true) => { return Err(format!( - "step {index}: import failed but should succeed (slashable data: {}): {}", - step.contains_slashable_data, - imported - .err() - .map(|err| err.to_string()) - .unwrap_or_default() + "step {index}: import failed but should succeed (slashable data: {}): \ + {refusal}", + step.contains_slashable_data )); } - (true, false) => { + (Ok(_), false) => { return Err(format!("step {index}: import succeeded but should fail")); } } @@ -190,14 +192,23 @@ fn run(case: &Case) -> Result<(), String> { Ok(()) } +/// What the importer says when it refuses an interchange for an attestation +/// whose source is at or past its target (`slashing_protection::file`). +const IMPOSSIBLE_SOURCE_REFUSAL: &str = "is at or after its target epoch"; + /// Whether `interchange` holds an attestation whose source is at or past its /// target outside genesis: the one shape this importer refuses rather than /// absorbs. The rule is restated here rather than borrowed from the crate, so /// that the importer widening what it refuses cannot widen what this runner /// forgives along with it. fn holds_impossible_source(interchange: &serde_json::Value) -> bool { - let epoch = - |value: &serde_json::Value| value.as_str().and_then(|text| text.parse::().ok()); + // Quoted, as EIP-3076 writes them and the vectors do, or bare, which the + // importer reads too. + let epoch = |value: &serde_json::Value| { + value + .as_u64() + .or_else(|| value.as_str().and_then(|text| text.parse().ok())) + }; interchange["data"] .as_array() .into_iter()