Fix handling of cross-fork messages in op pool

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2041,7 +2041,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn verify_voluntary_exit_for_gossip(
         &self,
         exit: SignedVoluntaryExit,
-    ) -> Result<ObservationOutcome<SignedVoluntaryExit>, Error> {
+    ) -> Result<ObservationOutcome<SignedVoluntaryExit, T::EthSpec>, Error> {
         // NOTE: this could be more efficient if it avoided cloning the head state
         let wall_clock_state = self.wall_clock_state()?;
         Ok(self
@@ -2062,7 +2062,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     }
 
     /// Accept a pre-verified exit and queue it for inclusion in an appropriate block.
-    pub fn import_voluntary_exit(&self, exit: SigVerifiedOp<SignedVoluntaryExit>) {
+    pub fn import_voluntary_exit(&self, exit: SigVerifiedOp<SignedVoluntaryExit, T::EthSpec>) {
         if self.eth1_chain.is_some() {
             self.op_pool.insert_voluntary_exit(exit)
         }
@@ -2072,7 +2072,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn verify_proposer_slashing_for_gossip(
         &self,
         proposer_slashing: ProposerSlashing,
-    ) -> Result<ObservationOutcome<ProposerSlashing>, Error> {
+    ) -> Result<ObservationOutcome<ProposerSlashing, T::EthSpec>, Error> {
         let wall_clock_state = self.wall_clock_state()?;
         Ok(self.observed_proposer_slashings.lock().verify_and_observe(
             proposer_slashing,
@@ -2082,7 +2082,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     }
 
     /// Accept some proposer slashing and queue it for inclusion in an appropriate block.
-    pub fn import_proposer_slashing(&self, proposer_slashing: SigVerifiedOp<ProposerSlashing>) {
+    pub fn import_proposer_slashing(
+        &self,
+        proposer_slashing: SigVerifiedOp<ProposerSlashing, T::EthSpec>,
+    ) {
         if self.eth1_chain.is_some() {
             self.op_pool.insert_proposer_slashing(proposer_slashing)
         }
@@ -2092,7 +2095,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn verify_attester_slashing_for_gossip(
         &self,
         attester_slashing: AttesterSlashing<T::EthSpec>,
-    ) -> Result<ObservationOutcome<AttesterSlashing<T::EthSpec>>, Error> {
+    ) -> Result<ObservationOutcome<AttesterSlashing<T::EthSpec>, T::EthSpec>, Error> {
         let wall_clock_state = self.wall_clock_state()?;
         Ok(self.observed_attester_slashings.lock().verify_and_observe(
             attester_slashing,
@@ -2107,7 +2110,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// 2. Add it to the op pool.
     pub fn import_attester_slashing(
         &self,
-        attester_slashing: SigVerifiedOp<AttesterSlashing<T::EthSpec>>,
+        attester_slashing: SigVerifiedOp<AttesterSlashing<T::EthSpec>, T::EthSpec>,
     ) {
         // Add to fork choice.
         self.canonical_head
@@ -2116,10 +2119,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // Add to the op pool (if we have the ability to propose blocks).
         if self.eth1_chain.is_some() {
-            self.op_pool.insert_attester_slashing(
-                attester_slashing,
-                self.canonical_head.cached_head().head_fork(),
-            )
+            self.op_pool.insert_attester_slashing(attester_slashing)
         }
     }
 

beacon_node/beacon_chain/src/observed_operations.rs

@@ -1,5 +1,6 @@
 use derivative::Derivative;
 use smallvec::SmallVec;
+use ssz::{Decode, Encode};
 use state_processing::{SigVerifiedOp, VerifyOperation};
 use std::collections::HashSet;
 use std::marker::PhantomData;
@@ -29,8 +30,8 @@ pub struct ObservedOperations<T: ObservableOperation<E>, E: EthSpec> {
 
 /// Was the observed operation new and valid for further processing, or a useless duplicate?
 #[derive(Debug, PartialEq, Eq, Clone)]
-pub enum ObservationOutcome<T> {
-    New(SigVerifiedOp<T>),
+pub enum ObservationOutcome<T: Encode + Decode, E: EthSpec> {
+    New(SigVerifiedOp<T, E>),
     AlreadyKnown,
 }
 
@@ -81,7 +82,7 @@ impl<T: ObservableOperation<E>, E: EthSpec> ObservedOperations<T, E> {
         op: T,
         head_state: &BeaconState<E>,
         spec: &ChainSpec,
-    ) -> Result<ObservationOutcome<T>, T::Error> {
+    ) -> Result<ObservationOutcome<T, E>, T::Error> {
         let observed_validator_indices = &mut self.observed_validator_indices;
         let new_validator_indices = op.observed_validators();
 
@@ -95,6 +96,8 @@ impl<T: ObservableOperation<E>, E: EthSpec> ObservedOperations<T, E> {
             .iter()
             .all(|index| observed_validator_indices.contains(index))
         {
+            // FIXME(sproul): consider verifying that those already-observed slashings are
+            // still valid.
             return Ok(ObservationOutcome::AlreadyKnown);
         }
 

beacon_node/beacon_chain/src/schema_change/migration_schema_v12.rs

@@ -23,12 +23,13 @@ pub fn upgrade_to_v12<T: BeaconChainTypes>(
         "count" => v5.attestations_v5.len(),
     );
 
+    // FIXME(sproul): actually transfer the operations across
     let v12 = PersistedOperationPool::V12(PersistedOperationPoolV12 {
         attestations: vec![],
         sync_contributions: v5.sync_contributions,
-        attester_slashings: v5.attester_slashings,
-        proposer_slashings: v5.proposer_slashings,
-        voluntary_exits: v5.voluntary_exits,
+        attester_slashings: vec![],
+        proposer_slashings: vec![],
+        voluntary_exits: vec![],
     });
     Ok(vec![v12.as_kv_store_op(OP_POOL_DB_KEY)])
 }
@@ -55,9 +56,9 @@ pub fn downgrade_from_v12<T: BeaconChainTypes>(
     let v5 = PersistedOperationPoolV5 {
         attestations_v5: vec![],
         sync_contributions: v12.sync_contributions,
-        attester_slashings: v12.attester_slashings,
-        proposer_slashings: v12.proposer_slashings,
-        voluntary_exits: v12.voluntary_exits,
+        attester_slashings_v5: vec![],
+        proposer_slashings_v5: vec![],
+        voluntary_exits_v5: vec![],
     };
     Ok(vec![v5.as_kv_store_op(OP_POOL_DB_KEY)])
 }

beacon_node/beacon_chain/src/test_utils.rs

@@ -1175,6 +1175,19 @@ where
     }
 
     pub fn make_attester_slashing(&self, validator_indices: Vec<u64>) -> AttesterSlashing<E> {
+        self.make_attester_slashing_with_epochs(validator_indices, None, None, None, None)
+    }
+
+    pub fn make_attester_slashing_with_epochs(
+        &self,
+        validator_indices: Vec<u64>,
+        source1: Option<Epoch>,
+        target1: Option<Epoch>,
+        source2: Option<Epoch>,
+        target2: Option<Epoch>,
+    ) -> AttesterSlashing<E> {
+        let fork = self.chain.canonical_head.cached_head().head_fork();
+
         let mut attestation_1 = IndexedAttestation {
             attesting_indices: VariableList::new(validator_indices).unwrap(),
             data: AttestationData {
@@ -1183,20 +1196,21 @@ where
                 beacon_block_root: Hash256::zero(),
                 target: Checkpoint {
                     root: Hash256::zero(),
-                    epoch: Epoch::new(0),
+                    epoch: target1.unwrap_or(fork.epoch),
                 },
                 source: Checkpoint {
                     root: Hash256::zero(),
-                    epoch: Epoch::new(0),
+                    epoch: source1.unwrap_or(Epoch::new(0)),
                 },
             },
             signature: AggregateSignature::infinity(),
         };
 
         let mut attestation_2 = attestation_1.clone();
         attestation_2.data.index += 1;
+        attestation_2.data.source.epoch = source2.unwrap_or(Epoch::new(0));
+        attestation_2.data.target.epoch = target2.unwrap_or(fork.epoch);
 
-        let fork = self.chain.canonical_head.cached_head().head_fork();
         for attestation in &mut [&mut attestation_1, &mut attestation_2] {
             for &i in &attestation.attesting_indices {
                 let sk = &self.validator_keypairs[i as usize].sk;
@@ -1280,8 +1294,19 @@ where
     }
 
     pub fn make_proposer_slashing(&self, validator_index: u64) -> ProposerSlashing {
+        self.make_proposer_slashing_at_slot(validator_index, None)
+    }
+
+    pub fn make_proposer_slashing_at_slot(
+        &self,
+        validator_index: u64,
+        slot_override: Option<Slot>,
+    ) -> ProposerSlashing {
         let mut block_header_1 = self.chain.head_beacon_block().message().block_header();
         block_header_1.proposer_index = validator_index;
+        if let Some(slot) = slot_override {
+            block_header_1.slot = slot;
+        }
 
         let mut block_header_2 = block_header_1.clone();
         block_header_2.state_root = Hash256::zero();

beacon_node/operation_pool/Cargo.toml

@@ -23,3 +23,4 @@ bitvec = "1"
 [dev-dependencies]
 beacon_chain =  { path = "../beacon_chain" }
 tokio = { version = "1.14.0", features = ["rt-multi-thread"] }
+maplit = "1.0.2"