chore(blend): drop proof generation from old epoch crypto processors (#3350)

blend/scheduling/src/message_blend/crypto/core_and_leader/mod.rs

@@ -1,2 +1,3 @@
+pub mod receive;
 pub mod send;
 pub mod send_and_receive;

blend/scheduling/src/message_blend/crypto/core_and_leader/receive.rs

@@ -0,0 +1,211 @@
+use lb_blend_message::{
+    Error,
+    crypto::proofs::PoQVerificationInputsMinusSigningKey,
+    encap::{
+        ProofsVerifier as ProofsVerifierTrait,
+        decapsulated::{DecapsulatedMessage, DecapsulationOutput},
+        encapsulated::EncapsulatedMessage,
+        validated::RequiredProofOfSelectionVerificationInputs,
+    },
+    reward::BlendingToken,
+};
+use lb_cryptarchia_engine::Epoch;
+use lb_key_management_system_keys::keys::X25519PrivateKey;
+
+use crate::{
+    membership::Membership, message_blend::crypto::EncapsulatedMessageWithVerifiedPublicHeader,
+};
+
+/// [`EpochCryptographicProcessor`] is responsible for only unwrapping the
+/// messages addressed to the local node.
+///
+/// Each instance is meant to be used during a single epoch.
+///
+/// It holds no proof generator, and hence cannot encapsulate anything:
+/// receiving spends no quota, so there is nothing for it to prove. That is what
+/// makes it the type for an epoch that has ended, which is kept around for the
+/// transition period only so that messages still in flight from it can be
+/// decapsulated and forwarded. Converting that epoch's send-and-receive
+/// processor with
+/// [`into_receiver_only`](super::send_and_receive::EpochCryptographicProcessor::into_receiver_only)
+/// drops its generators, and with them the `PoW` mining stream that would
+/// otherwise keep a core searching for solutions to a puzzle nobody will accept
+/// an answer to anymore.
+pub struct EpochCryptographicProcessor<ProofsVerifier> {
+    /// The non-ephemeral encryption key (NEK) for decapsulating messages.
+    non_ephemeral_encryption_key: X25519PrivateKey,
+    /// Index of the local node in the epoch's membership, `None` if the local
+    /// node is not a core node in it. The membership of an epoch does not
+    /// change, so this and the size below are resolved once, on construction.
+    local_node_index: Option<usize>,
+    membership_size: usize,
+    proofs_verifier: ProofsVerifier,
+    epoch: Epoch,
+}
+
+impl<ProofsVerifier> EpochCryptographicProcessor<ProofsVerifier> {
+    pub const fn verifier(&self) -> &ProofsVerifier {
+        &self.proofs_verifier
+    }
+
+    pub const fn epoch(&self) -> Epoch {
+        self.epoch
+    }
+}
+
+impl<ProofsVerifier> EpochCryptographicProcessor<ProofsVerifier>
+where
+    ProofsVerifier: ProofsVerifierTrait,
+{
+    #[must_use]
+    pub fn new<NodeId>(
+        non_ephemeral_encryption_key: X25519PrivateKey,
+        membership: &Membership<NodeId>,
+        public_info: PoQVerificationInputsMinusSigningKey,
+        epoch: Epoch,
+    ) -> Self {
+        Self {
+            non_ephemeral_encryption_key,
+            local_node_index: membership.local_index(),
+            membership_size: membership.size(),
+            proofs_verifier: ProofsVerifier::new(public_info),
+            epoch,
+        }
+    }
+
+    pub fn decapsulate_message(
+        &self,
+        message: EncapsulatedMessageWithVerifiedPublicHeader,
+    ) -> Result<DecapsulationOutput, Error> {
+        let Some(local_node_index) = self.local_node_index else {
+            return Err(Error::NotCoreNodeReceiver);
+        };
+        message.decapsulate(
+            &self.non_ephemeral_encryption_key,
+            &RequiredProofOfSelectionVerificationInputs {
+                expected_node_index: local_node_index as u64,
+                total_membership_size: self.membership_size as u64,
+            },
+            &self.proofs_verifier,
+        )
+    }
+
+    /// Validate the public header of an [`EncapsulatedMessage`].
+    pub fn validate_message_header(
+        &self,
+        message: EncapsulatedMessage,
+    ) -> Result<EncapsulatedMessageWithVerifiedPublicHeader, Error> {
+        message.verify_public_header(&self.proofs_verifier)
+    }
+
+    /// Semantically similar to [`Self::decapsulate_message`], but it does not
+    /// stop after decapsulating the outermost layer. It stops only when a layer
+    /// cannot be decapsulated or when the decapsulation is completed.
+    ///
+    /// If no layer (`Err`) or at most one layer (`Ok`) can be decapsulated,
+    /// this is semantically equivalent to calling
+    /// [`Self::decapsulate_message`].
+    ///
+    /// If more than a single layer can be decapsulated, then the decapsulation
+    /// happens recursively until the first layer that cannot be decapsulated is
+    /// found or when there is no more layers to decapsulate. In either case, it
+    /// returns the last processed layer, along with the list of blending tokens
+    /// collected along the way.
+    pub fn decapsulate_message_recursive(
+        &self,
+        message: EncapsulatedMessageWithVerifiedPublicHeader,
+    ) -> Result<MultiLayerDecapsulationOutput, Error> {
+        tracing::trace!(
+            "Attempt at batch-decapsulating message with PoQ nullifier and key: ({:?}, {:?})",
+            message.public_header().signing_key(),
+            message.public_header().proof_of_quota().key_nullifier()
+        );
+        let mut decapsulation_output = self.decapsulate_message(message)?;
+
+        let mut collected_blending_tokens = Vec::new();
+
+        loop {
+            match &decapsulation_output {
+                // We reached the end. Collect token and stop.
+                DecapsulationOutput::Completed { blending_token, .. } => {
+                    collected_blending_tokens.push(blending_token.clone());
+                    break;
+                }
+                // One or more layers to decapsulate. Collect token from current layer and attempt
+                // one more decapsulation.
+                DecapsulationOutput::Incompleted {
+                    remaining_encapsulated_message,
+                    blending_token,
+                } => {
+                    collected_blending_tokens.push(blending_token.clone());
+                    // If we find a message with an invalid public header after a successful
+                    // decapsulation, we still bubble it up for the scheduler to
+                    // schedule it. At the time of release, the message will be
+                    // ignored since its public header cannot be verified. This is not the most
+                    // efficient way, but it's the less invasive way since by decapsulation we
+                    // currently mean decrypting an encrypted Blend header. No additional checks are
+                    // performed on the nested public header. The spec simply ignores the message,
+                    // and so we do.
+                    let Ok(message_with_validated_public_header) =
+                        self.validate_message_header((**remaining_encapsulated_message).clone())
+                    else {
+                        break;
+                    };
+                    let Ok(nested_layer_decapsulation_output) =
+                        self.decapsulate_message(message_with_validated_public_header)
+                    else {
+                        break;
+                    };
+                    decapsulation_output = nested_layer_decapsulation_output;
+                }
+            }
+        }
+
+        Ok(MultiLayerDecapsulationOutput {
+            blending_tokens: collected_blending_tokens,
+            decapsulated_message: decapsulation_output.into(),
+        })
+    }
+}
+
+/// The output of a multi-layer decapsulation operation.
+#[derive(Debug)]
+pub struct MultiLayerDecapsulationOutput {
+    /// The blending token collected on the way, one per decapsulated layer.
+    blending_tokens: Vec<BlendingToken>,
+    /// The final message type.
+    decapsulated_message: DecapsulatedMessageType,
+}
+
+impl MultiLayerDecapsulationOutput {
+    #[must_use]
+    pub fn into_components(self) -> (Vec<BlendingToken>, DecapsulatedMessageType) {
+        (self.blending_tokens, self.decapsulated_message)
+    }
+}
+
+/// The final message type of a multi-layer decapsulation operation.
+#[derive(Debug)]
+pub enum DecapsulatedMessageType {
+    /// The remainder of the message still needs to be decapsulated by some
+    /// other node.
+    Incompleted(Box<EncapsulatedMessage>),
+    /// The message was fully decapsulated, as all the remaining encapsulations
+    /// were addressed to this node.
+    Completed(DecapsulatedMessage),
+}
+
+impl From<DecapsulationOutput> for DecapsulatedMessageType {
+    fn from(value: DecapsulationOutput) -> Self {
+        match value {
+            DecapsulationOutput::Completed {
+                fully_decapsulated_message,
+                ..
+            } => Self::Completed(fully_decapsulated_message),
+            DecapsulationOutput::Incompleted {
+                remaining_encapsulated_message,
+                ..
+            } => Self::Incompleted(remaining_encapsulated_message),
+        }
+    }
+}

blend/scheduling/src/message_blend/crypto/core_and_leader/send.rs

@@ -7,14 +7,11 @@ use lb_blend_message::{
 };
 use lb_cryptarchia_engine::Epoch;
 use lb_groth16::fr_to_bytes;
-use lb_key_management_system_keys::keys::X25519PrivateKey;
 
 use crate::{
     membership::Membership,
     message_blend::{
-        crypto::{
-            EncapsulatedMessageWithVerifiedPublicHeader, EpochCryptographicProcessorSettings,
-        },
+        crypto::EncapsulatedMessageWithVerifiedPublicHeader,
         provers::{
             BlendLayerProof, ProofsGeneratorSettings, WinningPolInfoStream,
             core_leader_and_pow::CoreLeaderAndPowProofsGenerator,
@@ -28,8 +25,6 @@ use crate::{
 /// Each instance is meant to be used during a single epoch.
 pub struct EpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator> {
     num_blend_layers: NonZeroU64,
-    /// The non-ephemeral encryption key (NEK) for decapsulating messages.
-    non_ephemeral_encryption_key: X25519PrivateKey,
     membership: Membership<NodeId>,
     proofs_generator: ProofsGenerator,
     _phantom: PhantomData<CorePoQGenerator>,
@@ -38,14 +33,6 @@ pub struct EpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator
 impl<NodeId, CorePoQGenerator, ProofsGenerator>
     EpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator>
 {
-    pub(super) const fn non_ephemeral_encryption_key(&self) -> &X25519PrivateKey {
-        &self.non_ephemeral_encryption_key
-    }
-
-    pub(super) const fn membership(&self) -> &Membership<NodeId> {
-        &self.membership
-    }
-
     #[cfg(test)]
     pub const fn proofs_generator(&self) -> &ProofsGenerator {
         &self.proofs_generator
@@ -64,7 +51,7 @@ where
 {
     #[must_use]
     pub fn new(
-        settings: EpochCryptographicProcessorSettings,
+        encapsulation_layers: NonZeroU64,
         membership: Membership<NodeId>,
         public_info: PoQVerificationInputsMinusSigningKey,
         core_proof_of_quota_generator: CorePoQGenerator,
@@ -78,12 +65,11 @@ where
             local_node_index: membership.local_index(),
             membership_size: membership.size(),
             public_inputs: public_info,
-            encapsulation_layers: settings.num_blend_layers,
+            encapsulation_layers,
             epoch,
         };
         Self {
-            num_blend_layers: settings.num_blend_layers,
-            non_ephemeral_encryption_key: settings.non_ephemeral_encryption_key,
+            num_blend_layers: encapsulation_layers,
             membership,
             proofs_generator: ProofsGenerator::new(
                 generator_settings,
@@ -101,11 +87,6 @@ where
         self.proofs_generator
             .set_epoch_private(winning_pol_info_stream, target_epoch);
     }
-
-    /// Stop generating proofs for this processor's epoch.
-    pub fn stop_proof_generation(&mut self) {
-        self.proofs_generator.drop_pow_proofs_stream();
-    }
 }
 
 impl<NodeId, CorePoQGenerator, ProofsGenerator>
@@ -238,9 +219,8 @@ mod test {
     use super::EpochCryptographicProcessor;
     use crate::{
         membership::{Membership, Node},
-        message_blend::crypto::{
-            EpochCryptographicProcessorSettings,
-            test_utils::{MockCorePoQGenerator, TestEpochChangeCoreAndLeaderProofsGenerator},
+        message_blend::crypto::test_utils::{
+            MockCorePoQGenerator, TestEpochChangeCoreAndLeaderProofsGenerator,
         },
     };
 
@@ -255,10 +235,7 @@ mod test {
         };
         let mut processor =
             EpochCryptographicProcessor::<_, _, TestEpochChangeCoreAndLeaderProofsGenerator>::new(
-                EpochCryptographicProcessorSettings {
-                    non_ephemeral_encryption_key: [0; _].into(),
-                    num_blend_layers: NonZeroU64::new(1).unwrap(),
-                },
+                NonZeroU64::new(1).unwrap(),
                 Membership::new_without_local(&[Node {
                     address: Multiaddr::empty(),
                     id: PeerId::random(),

blend/scheduling/src/message_blend/crypto/core_and_leader/send_and_receive.rs

@@ -3,10 +3,7 @@ use core::ops::{Deref, DerefMut};
 use lb_blend_message::{
     Error,
     crypto::proofs::PoQVerificationInputsMinusSigningKey,
-    encap::{
-        ProofsVerifier as ProofsVerifierTrait, decapsulated::DecapsulationOutput,
-        validated::RequiredProofOfSelectionVerificationInputs,
-    },
+    encap::{ProofsVerifier as ProofsVerifierTrait, decapsulated::DecapsulationOutput},
 };
 use lb_cryptarchia_engine::Epoch;
 
@@ -15,7 +12,10 @@ use crate::{
     message_blend::{
         crypto::{
             EncapsulatedMessageWithVerifiedPublicHeader, EpochCryptographicProcessorSettings,
-            core_and_leader::send::EpochCryptographicProcessor as SenderEpochCryptographicProcessor,
+            core_and_leader::{
+                receive::EpochCryptographicProcessor as ReceiverEpochCryptographicProcessor,
+                send::EpochCryptographicProcessor as SenderEpochCryptographicProcessor,
+            },
         },
         provers::core_leader_and_pow::CoreLeaderAndPowProofsGenerator,
     },
@@ -29,8 +29,7 @@ use crate::{
 /// This processor is suitable for core nodes.
 pub struct EpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier> {
     sender_processor: SenderEpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator>,
-    proofs_verifier: ProofsVerifier,
-    epoch: Epoch,
+    receiver_processor: ReceiverEpochCryptographicProcessor<ProofsVerifier>,
 }
 
 impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
@@ -39,12 +38,6 @@ where
     ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator>,
     ProofsVerifier: ProofsVerifierTrait,
 {
-    /// Stop generating proofs for this processor's epoch, while leaving it
-    /// able to decapsulate messages that are still in flight from it.
-    pub fn stop_proof_generation(&mut self) {
-        self.sender_processor.stop_proof_generation();
-    }
-
     #[must_use]
     pub fn new(
         settings: EpochCryptographicProcessorSettings,
@@ -53,16 +46,24 @@ where
         core_proof_of_quota_generator: CorePoQGenerator,
         epoch: Epoch,
     ) -> Self {
+        let EpochCryptographicProcessorSettings {
+            non_ephemeral_encryption_key,
+            num_blend_layers,
+        } = settings;
         Self {
+            receiver_processor: ReceiverEpochCryptographicProcessor::new(
+                non_ephemeral_encryption_key,
+                &membership,
+                public_info,
+                epoch,
+            ),
             sender_processor: SenderEpochCryptographicProcessor::new(
-                settings,
+                num_blend_layers,
                 membership,
                 public_info,
                 core_proof_of_quota_generator,
                 epoch,
             ),
-            proofs_verifier: ProofsVerifier::new(public_info),
-            epoch,
         }
     }
 }
@@ -71,11 +72,22 @@ impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
     EpochCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
 {
     pub const fn verifier(&self) -> &ProofsVerifier {
-        &self.proofs_verifier
+        self.receiver_processor.verifier()
     }
 
     pub const fn epoch(&self) -> Epoch {
-        self.epoch
+        self.receiver_processor.epoch()
+    }
+
+    pub const fn receiver(&self) -> &ReceiverEpochCryptographicProcessor<ProofsVerifier> {
+        &self.receiver_processor
+    }
+
+    /// Give up the send side of this processor, keeping only what it takes to
+    /// decapsulate.
+    #[must_use]
+    pub fn into_receiver_only(self) -> ReceiverEpochCryptographicProcessor<ProofsVerifier> {
+        self.receiver_processor
     }
 }
 
@@ -88,17 +100,7 @@ where
         &self,
         message: EncapsulatedMessageWithVerifiedPublicHeader,
     ) -> Result<DecapsulationOutput, Error> {
-        let Some(local_node_index) = self.sender_processor.membership().local_index() else {
-            return Err(Error::NotCoreNodeReceiver);
-        };
-        message.decapsulate(
-            self.sender_processor.non_ephemeral_encryption_key(),
-            &RequiredProofOfSelectionVerificationInputs {
-                expected_node_index: local_node_index as u64,
-                total_membership_size: self.sender_processor.membership().size() as u64,
-            },
-            &self.proofs_verifier,
-        )
+        self.receiver_processor.decapsulate_message(message)
     }
 }
 

blend/scheduling/src/message_blend/crypto/mod.rs

@@ -15,6 +15,7 @@ use lb_key_management_system_keys::keys::X25519PrivateKey;
 
 pub mod core_and_leader;
 pub use self::core_and_leader::{
+    receive::EpochCryptographicProcessor as CoreAndLeaderReceiverOnlyEpochCryptographicProcessor,
     send::EpochCryptographicProcessor as CoreAndLeaderSenderOnlyEpochCryptographicProcessor,
     send_and_receive::EpochCryptographicProcessor as CoreAndLeaderSendAndReceiveEpochCryptographicProcessor,
 };

blend/scheduling/src/message_blend/crypto/test_utils.rs

@@ -60,8 +60,6 @@ impl<CorePoQGenerator> CoreLeaderAndPowProofsGenerator<CorePoQGenerator>
         self.0 = Some(winning_pol_info_stream);
     }
 
-    fn drop_pow_proofs_stream(&mut self) {}
-
     async fn get_next_core_proof(&mut self) -> Option<BlendLayerProof> {
         None
     }

blend/scheduling/src/message_blend/provers/core_leader_and_pow/mod.rs

@@ -50,20 +50,11 @@ pub trait CoreLeaderAndPowProofsGenerator<CorePoQGenerator>: Sized {
     /// Request a new proof of work backed proof from the prover. It returns
     /// `None` if the epoch's `PoW` public inputs admit no proof at all.
     async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof>;
-    /// Stop the background work this generator is performing for its epoch.
-    ///
-    /// Called on the outgoing generator at an epoch rotation: it stays alive
-    /// through the transition period to verify messages still in flight, but
-    /// must not go on mining for an epoch that has ended.
-    fn drop_pow_proofs_stream(&mut self);
 }
 
 pub struct RealCoreLeaderAndPowProofsGenerator<CorePoQGenerator> {
     core_and_leader_proofs_generator: RealCoreAndLeaderProofsGenerator<CorePoQGenerator>,
-    /// `None` once generation has been stopped for this epoch. Dropping the
-    /// generator drops the mining stream it owns, which is what actually
-    /// abandons the work — see [`Self::stop_proof_generation`].
-    pow_proofs_generator: Option<RealPowProofsGenerator>,
+    pow_proofs_generator: RealPowProofsGenerator,
 }
 
 #[async_trait]
@@ -84,7 +75,7 @@ where
             // The `PoW` branch depends only on public epoch information, so
             // unlike the leadership branch it is ready from the moment the
             // generator is created.
-            pow_proofs_generator: Some(RealPowProofsGenerator::new(settings)),
+            pow_proofs_generator: RealPowProofsGenerator::new(settings),
         }
     }
 
@@ -109,16 +100,8 @@ where
             .await
     }
 
-    fn drop_pow_proofs_stream(&mut self) {
-        if self.pow_proofs_generator.take().is_some() {
-            tracing::debug!(target: LOG_TARGET, "Stopped PoW proof generation for this epoch.");
-        }
-    }
-
     async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof> {
-        // `None` once generation has been stopped for this epoch, which reads
-        // the same to a caller as an epoch whose `PoW` inputs admit no proof.
-        let generator = self.pow_proofs_generator.as_mut()?;
+        let generator = &mut self.pow_proofs_generator;
         let proof = generator.get_next_proof().await?;
         tracing::trace!(
             target: LOG_TARGET,

services/blend/src/core/mod.rs

@@ -33,7 +33,12 @@ use lb_blend::{
         EpochMessageScheduler,
         epoch::{EpochEvent, UninitializedEpochEventStream},
         message_blend::{
-            crypto::EpochCryptographicProcessorSettings,
+            crypto::{
+                EpochCryptographicProcessorSettings,
+                core_and_leader::receive::{
+                    DecapsulatedMessageType, MultiLayerDecapsulationOutput,
+                },
+            },
             provers::core_leader_and_pow::CoreLeaderAndPowProofsGenerator,
         },
         message_scheduler::{
@@ -78,8 +83,8 @@ use crate::{
         backends::BackendEpochInfo,
         kms::{KmsPoQAdapter, PreloadKMSBackendCorePoQGenerator},
         processor::{
-            CoreCryptographicProcessor, DecapsulatedMessageType, Error,
-            MultiLayerDecapsulationOutput,
+            CoreCryptographicProcessor as CurrentEpochCryptographicProcessor, Error,
+            ReceiverCryptographicProcessor,
         },
         scheduler::SchedulerWrapper,
         settings::{RunningBlendConfig, StartingBlendConfig},
@@ -108,6 +113,9 @@ pub use state::RecoveryServiceState as CoreServiceState;
 
 const LOG_TARGET: &str = blend::service::CORE;
 
+type OldEpochCryptographicProcessor<ProofsVerifier> =
+    ReceiverCryptographicProcessor<ProofsVerifier>;
+
 /// A blend service that sends messages to the blend network
 /// and broadcasts fully unwrapped messages through the [`NetworkService`].
 ///
@@ -495,7 +503,7 @@ async fn initialize<
     + Send
     + 'static,
     CoreEpochPublicInfo<NodeId>,
-    CoreCryptographicProcessor<
+    CurrentEpochCryptographicProcessor<
         NodeId,
         KmsAdapter::CorePoQGenerator,
         ProofsGenerator,
@@ -607,7 +615,7 @@ where
         pow: current_epoch_public_info.poq_pow_public_inputs,
     };
 
-    let crypto_processor = CoreCryptographicProcessor::<
+    let crypto_processor = CurrentEpochCryptographicProcessor::<
         _,
         KmsAdapter::CorePoQGenerator,
         ProofsGenerator,
@@ -794,7 +802,7 @@ async fn run_event_loop<
     >,
     rng: &mut Rng,
     mut pending_transactions: VecDeque<Vec<u8>>,
-    mut crypto_processor: CoreCryptographicProcessor<
+    mut crypto_processor: CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -803,7 +811,7 @@ async fn run_event_loop<
     mut current_epoch_info: CoreEpochPublicInfo<NodeId>,
     mut recovery_checkpoint: ServiceState<Backend::Settings, Dispatcher::Settings>,
 ) -> (
-    CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
+    OldEpochCryptographicProcessor<ProofsVerifier>,
     OldEpochMessageScheduler<Rng, ProcessedMessage, EncapsulatedMessageWithVerifiedPublicHeader>,
     OldEpochBlendingTokenCollector,
 )
@@ -819,9 +827,8 @@ where
 {
     // An optional crypto processor to handle the old epoch during transition
     // period.
-    let mut old_epoch_crypto_processor: Option<
-        CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
-    > = None;
+    let mut old_epoch_crypto_processor: Option<OldEpochCryptographicProcessor<ProofsVerifier>> =
+        None;
     let mut old_epoch_message_scheduler: Option<
         OldEpochMessageScheduler<
             Rng,
@@ -835,7 +842,7 @@ where
         // `old_epoch` captured here so we can drop the `Sync` requirement.
         let old_epoch = old_epoch_crypto_processor
             .as_ref()
-            .map(CoreCryptographicProcessor::epoch);
+            .map(OldEpochCryptographicProcessor::epoch);
         tokio::select! {
             Some(msg) = inbound_relay.next() => {
                 match msg {
@@ -860,7 +867,7 @@ where
                 recovery_checkpoint = handle_local_transaction(&encapsulation, &mut pending_transactions, &crypto_processor, &mut message_scheduler, recovery_checkpoint);
             }
             Some(incoming_message) = blend_messages.next() => {
-                recovery_checkpoint = handle_incoming_blend_message(incoming_message, &mut message_scheduler, old_epoch_message_scheduler.as_mut(), &crypto_processor, old_epoch_crypto_processor.as_ref(),  recovery_checkpoint);
+                recovery_checkpoint = handle_incoming_blend_message(incoming_message, &mut message_scheduler, old_epoch_message_scheduler.as_mut(), crypto_processor.receiver(), old_epoch_crypto_processor.as_ref(),  recovery_checkpoint);
             }
             Some(round_info) = message_scheduler.next() => {
                 recovery_checkpoint = handle_release_round(round_info, &mut crypto_processor, rng, backend, payload_dispatcher, recovery_checkpoint).await;
@@ -956,7 +963,7 @@ where
 /// again.
 async fn encapsulate_next_transaction<NodeId, ProofsGenerator, ProofsVerifier, CorePoQGenerator>(
     pending_transactions: &VecDeque<Vec<u8>>,
-    cryptographic_processor: &mut CoreCryptographicProcessor<
+    cryptographic_processor: &mut CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -995,7 +1002,7 @@ fn handle_local_transaction<
 >(
     encapsulation: &EncapsulatedMessageWithVerifiedPublicHeader,
     pending_transactions: &mut VecDeque<Vec<u8>>,
-    cryptographic_processor: &CoreCryptographicProcessor<
+    cryptographic_processor: &CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -1037,7 +1044,6 @@ async fn retire<
     Backend,
     Rng,
     Dispatcher,
-    ProofsGenerator,
     ProofsVerifier,
     CorePoQGenerator,
     RuntimeServiceId,
@@ -1060,18 +1066,12 @@ async fn retire<
     >,
     mut rng: Rng,
     mut blending_token_collector: OldEpochBlendingTokenCollector,
-    crypto_processor: CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    crypto_processor: OldEpochCryptographicProcessor<ProofsVerifier>,
 ) where
     NodeId: Clone + Eq + Hash + Send + Sync + 'static,
     Rng: rand::Rng + Clone + Send + Unpin,
     Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Send + Sync,
     Dispatcher: PayloadDispatcher<RuntimeServiceId> + Send + Sync,
-    ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator> + Send,
     CorePoQGenerator: Send + Sync,
     ProofsVerifier: ProofsVerifierTrait + Send + Sync,
     RuntimeServiceId: Send + Sync,
@@ -1118,7 +1118,7 @@ async fn handle_epoch_event<
 >(
     event: EpochEvent<MaybeEmptyCoreEpochInfo<NodeId, CorePoQGenerator>>,
     settings: &RunningBlendConfig<Backend::Settings>,
-    current_cryptographic_processor: CoreCryptographicProcessor<
+    current_cryptographic_processor: CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -1156,10 +1156,10 @@ where
                 core_poq_generator: new_core_poq_generator,
                 public: new_epoch_info,
             } = *core_epoch_info;
-            // Once a new epoch starts, old epoch's PoW work is useless, so we drop the PoW
-            // proof generator for the epoch transition period.
-            let mut current_cryptographic_processor = current_cryptographic_processor;
-            current_cryptographic_processor.stop_proof_generation();
+            // Once a new epoch starts, the old epoch's proving is useless: retiring
+            // its processor into a receive-only one for the transition period drops
+            // the generators, and with them the `PoW` mining they have in flight.
+            let old_cryptographic_processor = current_cryptographic_processor.rotate_epoch();
             let (
                 _,
                 _,
@@ -1203,7 +1203,7 @@ where
             let Some(core_poq_generator) = new_core_poq_generator else {
                 tracing::info!(target: LOG_TARGET, "Local node is not part of new membership. Retiring from core.");
                 return HandleEpochEventOutput::Retiring {
-                    old_crypto_processor: current_cryptographic_processor,
+                    old_crypto_processor: old_cryptographic_processor,
                     old_scheduler: Box::new(
                         current_scheduler
                             .rotate_epoch(new_scheduler_epoch_info, settings.scheduler_settings())
@@ -1213,8 +1213,8 @@ where
                 };
             };
 
-            let new_processor: CoreCryptographicProcessor<_, _, _, ProofsVerifier> =
-                match CoreCryptographicProcessor::try_new_with_core_condition_check(
+            let new_processor: CurrentEpochCryptographicProcessor<_, _, _, ProofsVerifier> =
+                match CurrentEpochCryptographicProcessor::try_new_with_core_condition_check(
                     new_epoch_info.membership.clone(),
                     settings.minimum_network_size,
                     EpochCryptographicProcessorSettings {
@@ -1246,7 +1246,7 @@ where
                     Err(e @ (Error::LocalIsNotCoreNode | Error::NetworkIsTooSmall(_))) => {
                         tracing::info!(target: LOG_TARGET, "New membership does not satisfy the core node condition: {e:?}");
                         return HandleEpochEventOutput::Retiring {
-                            old_crypto_processor: current_cryptographic_processor,
+                            old_crypto_processor: old_cryptographic_processor,
                             old_scheduler: Box::new(
                                 current_scheduler
                                     .rotate_epoch(
@@ -1264,7 +1264,7 @@ where
                 .rotate_epoch(new_scheduler_epoch_info, settings.scheduler_settings());
             HandleEpochEventOutput::Transitioning {
                 new_crypto_processor: new_processor,
-                old_crypto_processor: current_cryptographic_processor,
+                old_crypto_processor: old_cryptographic_processor,
                 new_scheduler,
                 old_scheduler: Box::new(old_scheduler),
                 new_recovery_checkpoint: ServiceState::with_epoch(
@@ -1280,14 +1280,7 @@ where
         }
         EpochEvent::NewEpoch(MaybeEmptyCoreEpochInfo::Empty { epoch, epoch_nonce }) => {
             tracing::info!(target: LOG_TARGET, "New epoch event received, but no epoch info is available due to empty membership set.");
-            // Reduce the scope of the `mut` borrow to this block only.
-            let current_cryptographic_processor = {
-                // TODO: Change the cryptographic processor type so that proving is dropped
-                // automatically on new epochs for the old epoch.
-                let mut current_cryptographic_processor = current_cryptographic_processor;
-                current_cryptographic_processor.stop_proof_generation();
-                current_cryptographic_processor
-            };
+            let old_cryptographic_processor = current_cryptographic_processor.rotate_epoch();
             let (_, _, _, _, _, current_epoch_blending_token_collector, _, _) =
                 current_recovery_checkpoint.into_components();
             let new_reward_epoch_info = reward::EpochInfo::new(
@@ -1301,7 +1294,7 @@ where
             let (_, old_epoch_blending_token_collector) =
                 current_epoch_blending_token_collector.rotate_epoch(&new_reward_epoch_info);
             HandleEpochEventOutput::Retiring {
-                old_crypto_processor: current_cryptographic_processor,
+                old_crypto_processor: old_cryptographic_processor,
                 old_scheduler: Box::new(current_scheduler.consume()),
                 old_token_collector: old_epoch_blending_token_collector,
             }
@@ -1359,10 +1352,13 @@ enum HandleEpochEventOutput<
     CorePoQGenerator,
 > {
     Transitioning {
-        new_crypto_processor:
-            CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
-        old_crypto_processor:
-            CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
+        new_crypto_processor: CurrentEpochCryptographicProcessor<
+            NodeId,
+            CorePoQGenerator,
+            ProofsGenerator,
+            ProofsVerifier,
+        >,
+        old_crypto_processor: OldEpochCryptographicProcessor<ProofsVerifier>,
         new_scheduler: EpochMessageScheduler<
             Rng,
             ProcessedMessage,
@@ -1379,8 +1375,12 @@ enum HandleEpochEventOutput<
         new_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
     },
     TransitionCompleted {
-        current_crypto_processor:
-            CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
+        current_crypto_processor: CurrentEpochCryptographicProcessor<
+            NodeId,
+            CorePoQGenerator,
+            ProofsGenerator,
+            ProofsVerifier,
+        >,
         current_scheduler: EpochMessageScheduler<
             Rng,
             ProcessedMessage,
@@ -1390,8 +1390,7 @@ enum HandleEpochEventOutput<
         new_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
     },
     Retiring {
-        old_crypto_processor:
-            CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
+        old_crypto_processor: OldEpochCryptographicProcessor<ProofsVerifier>,
         old_scheduler: Box<
             OldEpochMessageScheduler<
                 Rng,
@@ -1425,7 +1424,7 @@ async fn handle_local_block_proposal<
 >(
     proposal: &[u8],
     data_replication_factor: u64,
-    cryptographic_processor: &mut CoreCryptographicProcessor<
+    cryptographic_processor: &mut CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -1488,7 +1487,7 @@ fn schedule_local_encapsulated_message<
     CorePoQGenerator,
 >(
     wrapped_message: &EncapsulatedMessageWithVerifiedPublicHeader,
-    cryptographic_processor: &CoreCryptographicProcessor<
+    cryptographic_processor: &CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -1512,8 +1511,9 @@ where
     // Before blending the data message, we try to peel off any outer layers that
     // are addressed to us. In this case, we collect the blending tokens and we
     // blend only the remaining layers.
-    let self_decapsulation_output =
-        cryptographic_processor.decapsulate_message_recursive(wrapped_message.clone());
+    let self_decapsulation_output = cryptographic_processor
+        .receiver()
+        .decapsulate_message_recursive(wrapped_message.clone());
 
     let Ok(multi_layer_decapsulation_output) = self_decapsulation_output else {
         // The outermost layer of the data message is not for us, hence we treat this as
@@ -1592,15 +1592,7 @@ where
 /// included, which is what gated it from being relayed to the rest of the
 /// network — so all that is left here is to decapsulate it with the current or
 /// old epoch's cryptographic processor, depending on the epoch it comes from.
-fn handle_incoming_blend_message<
-    NodeId,
-    Rng,
-    BackendSettings,
-    NetworkSettings,
-    ProofsGenerator,
-    ProofsVerifier,
-    CorePoQGenerator,
->(
+fn handle_incoming_blend_message<Rng, BackendSettings, NetworkSettings, ProofsVerifier>(
     (verified_message, epoch): (EncapsulatedMessageWithVerifiedPublicHeader, Epoch),
     scheduler: &mut EpochMessageScheduler<
         Rng,
@@ -1614,19 +1606,11 @@ fn handle_incoming_blend_message<
             EncapsulatedMessageWithVerifiedPublicHeader,
         >,
     >,
-    cryptographic_processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
-    old_epoch_cryptographic_processor: Option<
-        &CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
-    >,
+    cryptographic_processor: &ReceiverCryptographicProcessor<ProofsVerifier>,
+    old_epoch_cryptographic_processor: Option<&OldEpochCryptographicProcessor<ProofsVerifier>>,
     current_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
 ) -> ServiceState<BackendSettings, NetworkSettings>
 where
-    NodeId: 'static,
     Rng: RngCore + Clone + Send + Unpin,
     BackendSettings: Clone,
     ProofsVerifier: ProofsVerifierTrait,
@@ -1663,14 +1647,9 @@ where
 
 /// Attempts recursive decapsulation of a message whose `PoQ` has already been
 /// verified. Returns `None` if decapsulation fails (already logged).
-fn try_decapsulate<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>(
+fn try_decapsulate<ProofsVerifier>(
     message: EncapsulatedMessageWithVerifiedPublicHeader,
-    processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    processor: &ReceiverCryptographicProcessor<ProofsVerifier>,
     epoch: Epoch,
 ) -> Option<MultiLayerDecapsulationOutput>
 where
@@ -1691,28 +1670,16 @@ where
 
 /// Same as [`handle_incoming_blend_message`] but only tries with
 /// the old epoch crypto processor.
-fn handle_incoming_blend_message_from_old_epoch<
-    Rng,
-    NodeId,
-    ProofsGenerator,
-    ProofsVerifier,
-    CorePoQGenerator,
->(
+fn handle_incoming_blend_message_from_old_epoch<Rng, ProofsVerifier>(
     verified_message: EncapsulatedMessageWithVerifiedPublicHeader,
     scheduler: &mut OldEpochMessageScheduler<
         Rng,
         ProcessedMessage,
         EncapsulatedMessageWithVerifiedPublicHeader,
     >,
-    cryptographic_processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    cryptographic_processor: &OldEpochCryptographicProcessor<ProofsVerifier>,
     blending_token_collector: &mut OldEpochBlendingTokenCollector,
 ) where
-    NodeId: 'static,
     ProofsVerifier: ProofsVerifierTrait,
 {
     let Some(output) = try_decapsulate(
@@ -1738,9 +1705,6 @@ fn handle_decapsulated_incoming_message_from_current_epoch<
     Rng,
     BackendSettings,
     NetworkSettings,
-    NodeId,
-    CorePoQGenerator,
-    ProofsGenerator,
     ProofsVerifier,
 >(
     multi_layer_decapsulation_output: MultiLayerDecapsulationOutput,
@@ -1750,12 +1714,7 @@ fn handle_decapsulated_incoming_message_from_current_epoch<
         EncapsulatedMessageWithVerifiedPublicHeader,
     >,
     current_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
-    cryptographic_processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    cryptographic_processor: &ReceiverCryptographicProcessor<ProofsVerifier>,
 ) -> ServiceState<BackendSettings, NetworkSettings>
 where
     BackendSettings: Clone,
@@ -1792,9 +1751,6 @@ fn handle_decapsulated_incoming_message_from_old_epoch<
     Rng,
     BackendSettings,
     NetworkSettings,
-    NodeId,
-    CorePoQGenerator,
-    ProofsGenerator,
     ProofsVerifier,
 >(
     multi_layer_decapsulation_output: MultiLayerDecapsulationOutput,
@@ -1804,12 +1760,7 @@ fn handle_decapsulated_incoming_message_from_old_epoch<
         EncapsulatedMessageWithVerifiedPublicHeader,
     >,
     recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
-    old_cryptographic_processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    old_cryptographic_processor: &OldEpochCryptographicProcessor<ProofsVerifier>,
 ) -> ServiceState<BackendSettings, NetworkSettings>
 where
     BackendSettings: Clone,
@@ -1836,20 +1787,10 @@ where
     clippy::cognitive_complexity,
     reason = "TODO: address this in a dedicated refactor"
 )]
-fn schedule_decapsulated_incoming_message<
-    NodeId,
-    CorePoQGenerator,
-    ProofsGenerator,
-    ProofsVerifier,
->(
+fn schedule_decapsulated_incoming_message<ProofsVerifier>(
     multi_layer_decapsulation_output: MultiLayerDecapsulationOutput,
     scheduler: &mut impl ProcessedMessageScheduler<ProcessedMessage>,
-    cryptographic_processor: &CoreCryptographicProcessor<
-        NodeId,
-        CorePoQGenerator,
-        ProofsGenerator,
-        ProofsVerifier,
-    >,
+    cryptographic_processor: &ReceiverCryptographicProcessor<ProofsVerifier>,
 ) -> (
     Option<ProcessedMessage>,
     impl Iterator<Item = BlendingToken>,
@@ -1932,7 +1873,7 @@ async fn handle_release_round<
         data_messages,
         release_type,
     }: RoundInfo<ProcessedMessage, EncapsulatedMessageWithVerifiedPublicHeader>,
-    cryptographic_processor: &mut CoreCryptographicProcessor<
+    cryptographic_processor: &mut CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -2166,7 +2107,7 @@ async fn generate_and_try_to_decapsulate_cover_message<
     ProofsVerifier,
     CorePoQGenerator,
 >(
-    cryptographic_processor: &mut CoreCryptographicProcessor<
+    cryptographic_processor: &mut CurrentEpochCryptographicProcessor<
         NodeId,
         CorePoQGenerator,
         ProofsGenerator,
@@ -2184,8 +2125,9 @@ where
         .encapsulate_cover_payload(&random_sized_bytes::<{ size_of::<u32>() }>())
         .await
         .expect("Should not fail to generate new cover message");
-    let self_decapsulation_output =
-        cryptographic_processor.decapsulate_message_recursive(encapsulated_cover_message.clone());
+    let self_decapsulation_output = cryptographic_processor
+        .receiver()
+        .decapsulate_message_recursive(encapsulated_cover_message.clone());
     let Ok(multi_layer_decapsulation_output) = self_decapsulation_output else {
         // First layer not addressed to ourselves. Publish as regular cover message,
         // hence we consume a core quota.

services/blend/src/core/processor.rs

@@ -4,17 +4,11 @@ use std::{
     ops::{Deref, DerefMut},
 };
 
+pub use lb_blend::scheduling::message_blend::crypto::core_and_leader::receive::EpochCryptographicProcessor as ReceiverCryptographicProcessor;
 use lb_blend::{
     message::{
-        Error as InnerError,
         crypto::proofs::PoQVerificationInputsMinusSigningKey,
-        encap::{
-            ProofsVerifier as ProofsVerifierTrait,
-            decapsulated::{DecapsulatedMessage, DecapsulationOutput},
-            encapsulated::EncapsulatedMessage,
-            validated::EncapsulatedMessageWithVerifiedPublicHeader,
-        },
-        reward::BlendingToken,
+        encap::ProofsVerifier as ProofsVerifierTrait,
     },
     scheduling::{
         membership::Membership,
@@ -39,22 +33,12 @@ impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
     pub const fn epoch(&self) -> Epoch {
         self.0.epoch()
     }
-}
 
-impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
-    CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
-where
-    ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator>,
-{
-    /// Stop generating proofs for this processor's epoch.
-    ///
-    /// The outgoing processor outlives its epoch by the transition period, so
-    /// that messages sent under the old public inputs can still be
-    /// decapsulated. Generating for that epoch is over as soon as the rotation
-    /// happens, and for the `PoW` branch that generation is a continuous
-    /// search that would otherwise keep a core busy for the whole period.
-    pub fn stop_proof_generation(&mut self) {
-        self.0.stop_proof_generation();
+    /// Retire this processor into the read-only one an epoch that has ended is
+    /// left with.
+    #[must_use]
+    pub fn rotate_epoch(self) -> ReceiverCryptographicProcessor<ProofsVerifier> {
+        self.0.into_receiver_only()
     }
 }
 
@@ -107,133 +91,6 @@ where
     }
 }
 
-/// The output of a multi-layer decapsulation operation.
-#[derive(Debug)]
-pub struct MultiLayerDecapsulationOutput {
-    /// The blending token collected on the way, one per decapsulated layer.
-    blending_tokens: Vec<BlendingToken>,
-    /// The final message type.
-    decapsulated_message: DecapsulatedMessageType,
-}
-
-impl MultiLayerDecapsulationOutput {
-    pub fn into_components(self) -> (Vec<BlendingToken>, DecapsulatedMessageType) {
-        (self.blending_tokens, self.decapsulated_message)
-    }
-}
-
-/// The final message type of a multi-layer decapsulation operation.
-#[derive(Debug)]
-pub enum DecapsulatedMessageType {
-    /// The remainder of the message still needs to be decapsulated by some
-    /// other node.
-    Incompleted(Box<EncapsulatedMessage>),
-    /// The message was fully decapsulated, as all the remaining encapsulations
-    /// were addressed to this node.
-    Completed(DecapsulatedMessage),
-}
-
-impl From<DecapsulationOutput> for DecapsulatedMessageType {
-    fn from(value: DecapsulationOutput) -> Self {
-        match value {
-            DecapsulationOutput::Completed {
-                fully_decapsulated_message,
-                ..
-            } => Self::Completed(fully_decapsulated_message),
-            DecapsulationOutput::Incompleted {
-                remaining_encapsulated_message,
-                ..
-            } => Self::Incompleted(remaining_encapsulated_message),
-        }
-    }
-}
-
-impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
-    CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
-where
-    ProofsVerifier: ProofsVerifierTrait,
-{
-    /// Validate the public header of an [`EncapsulatedMessage`].
-    pub fn validate_message_header(
-        &self,
-        message: EncapsulatedMessage,
-    ) -> Result<EncapsulatedMessageWithVerifiedPublicHeader, InnerError> {
-        message.verify_public_header(self.verifier())
-    }
-
-    /// Semantically similar to the underlying
-    /// [`EpochCryptographicProcessor::decapsulate_message`], but it does not
-    /// stop after decapsulating the outermost layer. It stops only when a layer
-    /// cannot be decapsulated or when the decapsulation is completed.
-    ///
-    /// If no layer (`Err`) or at most one layer (`Ok`) can be decapsulated,
-    /// this is semantically equivalent to
-    /// calling [`EpochCryptographicProcessor::decapsulate_message`].
-    ///
-    /// If more than a single layer can be decapsulated, then the decapsulation
-    /// happens recursively until the first layer that cannot be decapsulated is
-    /// found or when there is no more layers to decapsulate. In either case, it
-    /// returns the last processed layer, along with the list of blending tokens
-    /// collected along the way.
-    pub fn decapsulate_message_recursive(
-        &self,
-        message: EncapsulatedMessageWithVerifiedPublicHeader,
-    ) -> Result<MultiLayerDecapsulationOutput, InnerError> {
-        tracing::trace!(
-            "Attempt at batch-decapsulating message with PoQ nullifier and key: ({:?}, {:?})",
-            message.public_header().signing_key(),
-            message.public_header().proof_of_quota().key_nullifier()
-        );
-        let mut decapsulation_output = self.0.decapsulate_message(message)?;
-
-        let mut collected_blending_tokens = Vec::new();
-
-        loop {
-            match &decapsulation_output {
-                // We reached the end. Collect token and stop.
-                DecapsulationOutput::Completed { blending_token, .. } => {
-                    collected_blending_tokens.push(blending_token.clone());
-                    break;
-                }
-                // One or more layers to decapsulate. Collect token from current layer and attempt
-                // one more decapsulation.
-                DecapsulationOutput::Incompleted {
-                    remaining_encapsulated_message,
-                    blending_token,
-                } => {
-                    collected_blending_tokens.push(blending_token.clone());
-                    // If we find a message with an invalid public header after a successful
-                    // decapsulation, we still bubble it up for the scheduler to
-                    // schedule it. At the time of release, the message will be
-                    // ignored since its public header cannot be verified. This is not the most
-                    // efficient way, but it's the less invasive way since by decapsulation we
-                    // currently mean decrypting an encrypted Blend header. No additional checks are
-                    // performed on the nested public header. The spec simply ignores the message,
-                    // and so we do.
-                    let Ok(message_with_validated_public_header) = remaining_encapsulated_message
-                        .clone()
-                        .verify_public_header(self.verifier())
-                    else {
-                        break;
-                    };
-                    let Ok(nested_layer_decapsulation_output) = self
-                        .0
-                        .decapsulate_message(message_with_validated_public_header)
-                    else {
-                        break;
-                    };
-                    decapsulation_output = nested_layer_decapsulation_output;
-                }
-            }
-        }
-
-        Ok(MultiLayerDecapsulationOutput {
-            blending_tokens: collected_blending_tokens,
-            decapsulated_message: decapsulation_output.into(),
-        })
-    }
-}
-
 impl<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier> Deref
     for CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>
 {
@@ -281,7 +138,9 @@ mod tests {
             },
             selection::{self, VerifiedProofOfSelection},
         },
-        scheduling::message_blend::crypto::EpochCryptographicProcessorSettings,
+        scheduling::message_blend::crypto::{
+            EpochCryptographicProcessorSettings, core_and_leader::receive::DecapsulatedMessageType,
+        },
     };
     use lb_chain_service::Epoch;
     use lb_core::crypto::ZkHash;
@@ -290,7 +149,7 @@ mod tests {
     use lb_poq::Quota;
 
     use crate::{
-        core::processor::{CoreCryptographicProcessor, DecapsulatedMessageType, Error},
+        core::processor::{CoreCryptographicProcessor, Error},
         test_utils::{
             crypto::{MockCoreAndLeaderProofsGenerator, MockProofsVerifier, StaticFetchVerifier},
             membership::{key, membership},
@@ -395,7 +254,9 @@ mod tests {
             Epoch::new(0),
         );
         assert!(matches!(
-            processor.decapsulate_message_recursive(mock_message),
+            processor
+                .receiver()
+                .decapsulate_message_recursive(mock_message),
             Err(InnerError::ProofOfSelectionVerificationFailed(
                 selection::Error::Verification
             ))
@@ -427,6 +288,7 @@ mod tests {
         );
         StaticFetchVerifier::set_remaining_valid_poq_proofs(1);
         let decapsulation_output = processor
+            .receiver()
             .decapsulate_message_recursive(mock_message)
             .unwrap();
         let (blending_tokens, remaining_message_type) = decapsulation_output.into_components();
@@ -462,6 +324,7 @@ mod tests {
         );
         StaticFetchVerifier::set_remaining_valid_poq_proofs(2);
         let decapsulation_output = processor
+            .receiver()
             .decapsulate_message_recursive(mock_message)
             .unwrap();
         let (blending_tokens, remaining_message_type) = decapsulation_output.into_components();
@@ -497,6 +360,7 @@ mod tests {
         );
         StaticFetchVerifier::set_remaining_valid_poq_proofs(3);
         let decapsulation_output = processor
+            .receiver()
             .decapsulate_message_recursive(mock_message)
             .unwrap();
         let (blending_tokens, remaining_message_type) = decapsulation_output.into_components();

services/blend/src/core/tests/mod.rs

@@ -30,8 +30,7 @@ use crate::{
             TestPayloadDispatcher, backend_epoch_info, dummy_overwatch_resources,
             dummy_pol_private_inputs, new_crypto_processor, new_epoch_info, new_membership,
             new_stream, outgoing_messages_recorder, recorded_set_epoch_private_calls,
-            recorded_stop_proof_generation_calls, reset_set_epoch_private_calls,
-            reset_stop_proof_generation_calls, reward_epoch_info, scheduler_epoch_info,
+            reset_set_epoch_private_calls, reward_epoch_info, scheduler_epoch_info,
             scheduler_settings, sdp_relay, settings, timing_settings, wait_for_blend_backend_event,
         },
     },
@@ -116,7 +115,7 @@ async fn test_handle_incoming_blend_message() {
         (msg.clone(), 0.into()),
         &mut scheduler,
         None,
-        &processor,
+        processor.receiver(),
         None,
         recovery_checkpoint,
     );
@@ -130,7 +129,9 @@ async fn test_handle_incoming_blend_message() {
     );
 
     // Creates a new processor/scheduler/token_collector with the new epoch
-    // number.
+    // number. The outgoing processor is retired into its receive-only form,
+    // which is all it is good for during the transition period.
+    let processor = processor.rotate_epoch();
     epoch = epoch.strict_add(1.into());
     let public_info = new_epoch_info(epoch, membership.clone(), &settings);
     let mut new_processor = new_crypto_processor(
@@ -163,7 +164,7 @@ async fn test_handle_incoming_blend_message() {
         (msg.clone(), 0.into()),
         &mut new_scheduler,
         Some(&mut scheduler),
-        &new_processor,
+        new_processor.receiver(),
         Some(&processor),
         recovery_checkpoint,
     );
@@ -201,7 +202,7 @@ async fn test_handle_incoming_blend_message() {
         (msg, 1.into()),
         &mut new_scheduler,
         Some(&mut scheduler),
-        &new_processor,
+        new_processor.receiver(),
         Some(&processor),
         recovery_checkpoint,
     );
@@ -247,7 +248,7 @@ async fn test_handle_incoming_blend_message() {
         (msg, 2.into()),
         &mut new_scheduler,
         Some(&mut scheduler),
-        &new_processor,
+        new_processor.receiver(),
         Some(&processor),
         recovery_checkpoint,
     );
@@ -363,7 +364,7 @@ async fn test_duplicate_decapsulated_replica_handled_gracefully() {
         (replica_a, epoch),
         &mut scheduler,
         None,
-        &processor,
+        processor.receiver(),
         None,
         recovery_checkpoint,
     );
@@ -380,7 +381,7 @@ async fn test_duplicate_decapsulated_replica_handled_gracefully() {
         (replica_b, epoch),
         &mut scheduler,
         None,
-        &processor,
+        processor.receiver(),
         None,
         recovery_checkpoint,
     );
@@ -458,7 +459,7 @@ async fn test_handle_incoming_blend_message_with_invalid_poq() {
         (msg, epoch_1),
         &mut scheduler,
         None,
-        &processor_1,
+        processor_1.receiver(),
         None,
         recovery_checkpoint,
     ));
@@ -896,20 +897,6 @@ async fn transition_to_new_epoch_with_secret(secret_epoch: Epoch) -> Vec<Epoch>
     recorded_set_epoch_private_calls()
 }
 
-/// An epoch rotation must stop the outgoing epoch's proof generation.
-///
-/// The outgoing processor is kept for the transition period so messages still
-/// in flight from its epoch can be decapsulated, but its generators are done:
-/// a `PoW` solution is ground against one epoch's nonce and judged against
-/// that epoch's threshold, so mining for an epoch that has ended produces
-/// nothing usable while occupying a core for the whole period.
-#[test_log::test(tokio::test)]
-async fn test_handle_epoch_event_stops_old_epoch_proof_generation() {
-    reset_stop_proof_generation_calls();
-    let _calls = transition_to_new_epoch_with_secret(1.into()).await;
-    assert_eq!(recorded_stop_proof_generation_calls(), 1);
-}
-
 /// On an epoch change, if secret `PoL` info for the *new* epoch is already
 /// available (`current_secret_info`), it must be applied to the *new*
 /// cryptographic generator via `set_epoch_private`. If the available secret
@@ -1586,7 +1573,7 @@ async fn test_proof_generator_epoch_binding() {
         (msg_0.clone(), epoch_0),
         &mut scheduler_0,
         None,
-        &generator_0,
+        generator_0.receiver(),
         None,
         recovery_checkpoint,
     ));
@@ -1617,7 +1604,7 @@ async fn test_proof_generator_epoch_binding() {
         (msg_1.clone(), epoch_0),
         &mut scheduler_0_only,
         None,
-        &generator_0,
+        generator_0.receiver(),
         None,
         recovery_checkpoint,
     ));
@@ -1650,7 +1637,7 @@ async fn test_proof_generator_epoch_binding() {
         (msg_1, epoch_1),
         &mut scheduler_1,
         None,
-        &generator_1,
+        generator_1.receiver(),
         None,
         recovery_checkpoint,
     ));

services/blend/src/core/tests/utils.rs

@@ -453,25 +453,6 @@ thread_local! {
     static SET_EPOCH_PRIVATE_CALLS: RefCell<Vec<Epoch>> = const { RefCell::new(Vec::new()) };
 }
 
-thread_local! {
-    /// Counts the calls to
-    /// [`MockCoreAndLeaderProofsGenerator::stop_proof_generation`], so tests
-    /// can assert that an epoch rotation stops the outgoing epoch's proof
-    /// generation. Test-isolated for the same reason as above.
-    static STOP_PROOF_GENERATION_CALLS: RefCell<usize> = const { RefCell::new(0) };
-}
-
-/// Clears the count of `stop_proof_generation` calls.
-pub fn reset_stop_proof_generation_calls() {
-    STOP_PROOF_GENERATION_CALLS.with(|calls| *calls.borrow_mut() = 0);
-}
-
-/// How many times `stop_proof_generation` has been called since the last
-/// reset.
-pub fn recorded_stop_proof_generation_calls() -> usize {
-    STOP_PROOF_GENERATION_CALLS.with(|calls| *calls.borrow())
-}
-
 /// Clears the record of `set_epoch_private` calls. Call before the code under
 /// test to isolate the calls of interest.
 pub fn reset_set_epoch_private_calls() {
@@ -501,10 +482,6 @@ impl<CorePoQGenerator> CoreLeaderAndPowProofsGenerator<CorePoQGenerator>
         SET_EPOCH_PRIVATE_CALLS.with(|calls| calls.borrow_mut().push(target_epoch));
     }
 
-    fn drop_pow_proofs_stream(&mut self) {
-        STOP_PROOF_GENERATION_CALLS.with(|calls| *calls.borrow_mut() += 1);
-    }
-
     async fn get_next_core_proof(&mut self) -> Option<BlendLayerProof> {
         Some(epoch_based_dummy_proofs(self.0))
     }

services/blend/src/test_utils/crypto.rs

@@ -42,8 +42,6 @@ impl<CorePoQGenerator> CoreLeaderAndPowProofsGenerator<CorePoQGenerator>
     ) {
     }
 
-    fn drop_pow_proofs_stream(&mut self) {}
-
     async fn get_next_core_proof(&mut self) -> Option<BlendLayerProof> {
         Some(mock_blend_proof())
     }
@@ -198,8 +196,6 @@ impl<CorePoQGenerator> CoreLeaderAndPowProofsGenerator<CorePoQGenerator>
     ) {
     }
 
-    fn drop_pow_proofs_stream(&mut self) {}
-
     async fn get_next_core_proof(&mut self) -> Option<BlendLayerProof> {
         Some(mock_blend_proof())
     }