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 5372b8470..a857db8b2 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 @@ -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 @@ -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, @@ -763,6 +771,49 @@ 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. + /// + /// 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(); + 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) + .without_attestation_data_check(), + ); + 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..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,13 +502,37 @@ 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"); 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() { + // 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); + let right = MockBeaconNode::new().with_attestation_data(96); + let fallback = FallbackBeaconNode::new(vec![wrong, right]); + + let data = fallback + .attestation_data(96, fork_of(96)) + .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..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, &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 49c1c54dd..30a813fce 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,28 @@ 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`). +/// +/// `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!( "requested attestation data for slot {slot}, node answered for slot {}", @@ -74,9 +90,54 @@ 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) -> u64 { + 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 => 0, + // EIP-7549 moved the committee out of the signed data and left the + // field fixed at zero. + 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 => 1, + // Not a Beacon Chain fork: `Config::fork_at_epoch` never names it, so + // 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" + ), + } +} + /// The chain's genesis, as the beacon node reports it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Genesis { @@ -339,6 +400,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 +676,76 @@ 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 and no schedule names + /// it, so a caller that does has a bug, reported as one rather than as a + /// node answering badly. + #[test] + #[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/error.rs b/crates/validator/src/error.rs index ae65aee92..33ec4f3d5 100644 --- a/crates/validator/src/error.rs +++ b/crates/validator/src/error.rs @@ -115,15 +115,18 @@ 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. 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 c8356decb..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)) => { - 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!( - "keystore rewritten, but the slashing-protection record could not be \ - written: {err}; {entry}" - )) - } else if rolled_back.is_err() { - // Still on disk, not loaded now, and refused at the next - // start until it has a record. - 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}" - )) - } - } + (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. @@ -1376,7 +1462,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 +1506,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 +1579,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()), @@ -1508,11 +1594,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"); @@ -1562,7 +1649,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()), @@ -1637,9 +1724,174 @@ mod tests { assert_eq!(definitions.0, vec![previous]); } - /// A request that carries history is refused whole, importing nothing. + /// 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 + /// `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 +1910,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()) @@ -1682,7 +1942,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()), @@ -1720,6 +1980,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. @@ -1740,7 +2112,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( @@ -1777,7 +2149,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 24fc99dfd..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,117 +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); - - 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 @@ -1411,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/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..a76f7ea0e 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,11 +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 panic that poisons -/// it is reported the same way on the call that raised it, which sees it as -/// the blocking task's `JoinError`. +/// 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 @@ -113,13 +148,12 @@ 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(); + 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 @@ -127,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()), + 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 { @@ -148,16 +185,43 @@ pub async fn blocking( }) .await .map_err(|err| { - if err.is_panic() { - poisoned() + 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, 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 { - Error::SlashingProtectionUnusable { - reason: format!("the record could not be consulted: {err}"), - } - } + 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 no attestation or block \ + is signed until the validator client is restarted" + .to_string(), + } +} + /// 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, @@ -529,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. @@ -874,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 { @@ -1459,6 +1525,35 @@ 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 _; + + 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"); + 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"); @@ -1517,4 +1612,100 @@ mod tests { Err(Refusal::Unregistered) ); } + + // A poisoned record. + + /// 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 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"); + panic!("a bug while holding the record"); + }) + .join() + .expect_err("the thread panics"); + } + + /// 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_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"); + 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}"); + } + + /// 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/crates/validator/tests/interchange_vectors.rs b/crates/validator/tests/interchange_vectors.rs index a338d8b73..19d6e11c9 100644 --- a/crates/validator/tests/interchange_vectors.rs +++ b/crates/validator/tests/interchange_vectors.rs @@ -122,26 +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 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(()), - (false, true) => { + // 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. + (Err(refusal), true) + if step.contains_slashable_data + && holds_impossible_source(&step.interchange) + && refusal.contains(IMPOSSIBLE_SOURCE_REFUSAL) => + { + return Ok(()); + } + (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")); } } @@ -182,6 +192,35 @@ 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 { + // 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() + .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..64c4ac36e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -277,7 +277,13 @@ 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 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 @@ -311,8 +317,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. @@ -474,7 +484,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